332 lines
11 KiB
Rust
332 lines
11 KiB
Rust
//! A fake Mattermost on 127.0.0.1, plain TCP: the four REST calls `gatewayd` makes, and the
|
|
//! WebSocket, whose events the test sends and whose requests it reads. Do not edit.
|
|
|
|
#![allow(dead_code)] // each test file uses a different part of this module
|
|
|
|
use std::collections::HashMap;
|
|
use std::io::{Read, Write};
|
|
use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream};
|
|
use std::sync::{Arc, Mutex, mpsc};
|
|
use std::time::{Duration, Instant};
|
|
|
|
use gatewayd::ws::handshake::accept_for;
|
|
use serde_json::{Value, json};
|
|
|
|
pub const BOT: &str = "b0000000000000000000000000";
|
|
pub const BOT_NAME: &str = "boxmaker-straylight";
|
|
pub const KYLE: &str = "k0000000000000000000000000";
|
|
pub const EVE: &str = "e0000000000000000000000000";
|
|
/// The direct channel between the bot and Kyle, and between the bot and anyone else.
|
|
pub const DM: &str = "d0000000000000000000000000";
|
|
pub const EVE_DM: &str = "f0000000000000000000000000";
|
|
|
|
#[derive(Default)]
|
|
struct Inner {
|
|
/// Status for `users/me`: 200 unless a test sets another.
|
|
me_status: u16,
|
|
/// The body for `channels/<id>/posts?since=`, by channel.
|
|
since: HashMap<String, Value>,
|
|
/// Every post made: channel, root, message.
|
|
posts: Vec<(String, String, String)>,
|
|
/// Every REST call: method and path.
|
|
calls: Vec<(String, String)>,
|
|
}
|
|
|
|
pub struct FakeMm {
|
|
pub addr: SocketAddr,
|
|
inner: Arc<Mutex<Inner>>,
|
|
sockets: Mutex<mpsc::Receiver<WsPeer>>,
|
|
}
|
|
|
|
/// One WebSocket connection from `gatewayd`.
|
|
pub struct WsPeer {
|
|
writer: TcpStream,
|
|
/// The text of every text frame `gatewayd` sends.
|
|
pub texts: mpsc::Receiver<String>,
|
|
}
|
|
|
|
/// A post as Mattermost sends it.
|
|
pub fn post(
|
|
id: &str,
|
|
user: &str,
|
|
channel: &str,
|
|
root: &str,
|
|
message: &str,
|
|
create_at: i64,
|
|
) -> Value {
|
|
json!({
|
|
"id": id, "create_at": create_at, "update_at": create_at, "delete_at": 0, "user_id": user,
|
|
"channel_id": channel, "root_id": root, "message": message, "type": "", "props": {}
|
|
})
|
|
}
|
|
|
|
pub fn id(prefix: char, n: u32) -> String {
|
|
format!("{prefix}{n:025}")
|
|
}
|
|
|
|
fn frame(opcode: u8, payload: &[u8]) -> Vec<u8> {
|
|
let mut out = vec![0x80 | opcode];
|
|
match payload.len() {
|
|
n if n < 126 => out.push(n as u8),
|
|
n => {
|
|
out.push(126);
|
|
out.extend_from_slice(&(n as u16).to_be_bytes());
|
|
}
|
|
}
|
|
out.extend_from_slice(payload);
|
|
out
|
|
}
|
|
|
|
impl WsPeer {
|
|
pub fn event(&mut self, value: &Value) {
|
|
let _ = self
|
|
.writer
|
|
.write_all(&frame(0x1, value.to_string().as_bytes()));
|
|
}
|
|
|
|
pub fn posted(&mut self, post: &Value, channel_type: &str) {
|
|
let data = json!({"post": post.to_string(), "channel_type": channel_type, "team_id": ""});
|
|
self.event(&json!({"event": "posted", "data": data, "broadcast": {}, "seq": 1}));
|
|
}
|
|
|
|
/// End the connection without a close frame.
|
|
pub fn drop_connection(self) {
|
|
let _ = self.writer.shutdown(Shutdown::Both);
|
|
}
|
|
|
|
/// The `user_typing` requests received within `wait`, as (channel, parent).
|
|
pub fn typing_within(&self, wait: Duration) -> Vec<(String, String)> {
|
|
let until = Instant::now() + wait;
|
|
let mut got = Vec::new();
|
|
while let Ok(text) = self
|
|
.texts
|
|
.recv_timeout(until.saturating_duration_since(Instant::now()))
|
|
{
|
|
let v: Value = serde_json::from_str(&text).unwrap();
|
|
if v["action"] == "user_typing" {
|
|
let data = &v["data"];
|
|
got.push((
|
|
data["channel_id"].as_str().unwrap().to_string(),
|
|
data["parent_id"].as_str().unwrap().to_string(),
|
|
));
|
|
}
|
|
}
|
|
got
|
|
}
|
|
}
|
|
|
|
fn read_head(stream: &mut TcpStream) -> Option<String> {
|
|
let mut head = Vec::new();
|
|
let mut byte = [0u8; 1];
|
|
while !head.ends_with(b"\r\n\r\n") {
|
|
if stream.read(&mut byte).ok()? == 0 {
|
|
return None;
|
|
}
|
|
head.push(byte[0]);
|
|
}
|
|
String::from_utf8(head).ok()
|
|
}
|
|
|
|
/// Unmask the client's frames and send each text on `tx`, until the connection ends.
|
|
fn read_frames(mut stream: TcpStream, tx: mpsc::Sender<String>) {
|
|
let mut exact = |n: usize| -> Option<Vec<u8>> {
|
|
let mut buf = vec![0u8; n];
|
|
stream.read_exact(&mut buf).ok().map(|()| buf)
|
|
};
|
|
loop {
|
|
let Some(head) = exact(2) else { return };
|
|
let len = match head[1] & 0x7F {
|
|
126 => u16::from_be_bytes(exact(2).unwrap().try_into().unwrap()) as usize,
|
|
127 => return,
|
|
n => n as usize,
|
|
};
|
|
let Some(mask) = exact(4) else { return };
|
|
let Some(raw) = exact(len) else { return };
|
|
let payload: Vec<u8> = raw
|
|
.iter()
|
|
.zip(mask.iter().cycle())
|
|
.map(|(b, m)| b ^ m)
|
|
.collect();
|
|
if head[0] & 0x0F == 0x1 {
|
|
let _ = tx.send(String::from_utf8(payload).unwrap());
|
|
}
|
|
}
|
|
}
|
|
|
|
impl FakeMm {
|
|
pub fn start() -> FakeMm {
|
|
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
|
let addr = listener.local_addr().unwrap();
|
|
let inner = Arc::new(Mutex::new(Inner {
|
|
me_status: 200,
|
|
..Inner::default()
|
|
}));
|
|
let (ws_tx, ws_rx) = mpsc::channel();
|
|
let shared = Arc::clone(&inner);
|
|
std::thread::spawn(move || {
|
|
for stream in listener.incoming() {
|
|
let Ok(stream) = stream else { continue };
|
|
let (inner, ws_tx) = (Arc::clone(&shared), ws_tx.clone());
|
|
std::thread::spawn(move || connection(stream, &inner, &ws_tx));
|
|
}
|
|
});
|
|
FakeMm {
|
|
addr,
|
|
inner,
|
|
sockets: Mutex::new(ws_rx),
|
|
}
|
|
}
|
|
|
|
pub fn url(&self) -> String {
|
|
format!("http://127.0.0.1:{}", self.addr.port())
|
|
}
|
|
|
|
pub fn refuse_token(&self) {
|
|
self.inner.lock().unwrap().me_status = 401;
|
|
}
|
|
|
|
pub fn set_since(&self, channel: &str, posts: &[Value]) {
|
|
let order: Vec<Value> = posts.iter().map(|p| p["id"].clone()).collect();
|
|
let map: serde_json::Map<String, Value> = posts
|
|
.iter()
|
|
.map(|p| (p["id"].as_str().unwrap().to_string(), p.clone()))
|
|
.collect();
|
|
self.inner
|
|
.lock()
|
|
.unwrap()
|
|
.since
|
|
.insert(channel.to_string(), json!({"order": order, "posts": map}));
|
|
}
|
|
|
|
/// The next WebSocket `gatewayd` opens, after its hello.
|
|
pub fn next_ws(&self, wait: Duration) -> WsPeer {
|
|
self.sockets
|
|
.lock()
|
|
.unwrap()
|
|
.recv_timeout(wait)
|
|
.expect("no WebSocket connection")
|
|
}
|
|
|
|
pub fn posts(&self) -> Vec<(String, String, String)> {
|
|
self.inner.lock().unwrap().posts.clone()
|
|
}
|
|
|
|
/// Wait until at least `n` posts were made, for at most `wait`.
|
|
pub fn wait_posts(&self, n: usize, wait: Duration) -> Vec<(String, String, String)> {
|
|
let until = Instant::now() + wait;
|
|
while self.posts().len() < n && Instant::now() < until {
|
|
std::thread::sleep(Duration::from_millis(10));
|
|
}
|
|
self.posts()
|
|
}
|
|
|
|
pub fn calls(&self) -> Vec<(String, String)> {
|
|
self.inner.lock().unwrap().calls.clone()
|
|
}
|
|
}
|
|
|
|
fn connection(mut stream: TcpStream, inner: &Mutex<Inner>, ws_tx: &mpsc::Sender<WsPeer>) {
|
|
let Some(head) = read_head(&mut stream) else {
|
|
return;
|
|
};
|
|
let mut words = head.split_whitespace();
|
|
let (method, path) = (
|
|
words.next().unwrap_or("").to_string(),
|
|
words.next().unwrap_or("").to_string(),
|
|
);
|
|
if path == "/api/v4/websocket" {
|
|
let key = head
|
|
.lines()
|
|
.find_map(|l| l.strip_prefix("Sec-WebSocket-Key: "))
|
|
.unwrap_or("")
|
|
.trim()
|
|
.to_string();
|
|
let reply = format!(
|
|
"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {}\r\n\r\n",
|
|
accept_for(&key)
|
|
);
|
|
let _ = stream.write_all(reply.as_bytes());
|
|
let _ = stream.write_all(&frame(
|
|
0x1,
|
|
br#"{"event":"hello","data":{},"broadcast":{},"seq":0}"#,
|
|
));
|
|
let (tx, texts) = mpsc::channel();
|
|
let reader = stream.try_clone().unwrap();
|
|
std::thread::spawn(move || read_frames(reader, tx));
|
|
let _ = ws_tx.send(WsPeer {
|
|
writer: stream,
|
|
texts,
|
|
});
|
|
return;
|
|
}
|
|
let length = head
|
|
.lines()
|
|
.find_map(|l| {
|
|
l.to_ascii_lowercase()
|
|
.strip_prefix("content-length:")
|
|
.map(|v| v.trim().parse::<usize>().unwrap_or(0))
|
|
})
|
|
.unwrap_or(0);
|
|
let mut body = vec![0u8; length];
|
|
let _ = stream.read_exact(&mut body);
|
|
let (status, answer) = rest(inner, &method, &path, &body);
|
|
let text = answer.to_string();
|
|
let reply = format!(
|
|
"HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{text}",
|
|
text.len()
|
|
);
|
|
let _ = stream.write_all(reply.as_bytes());
|
|
}
|
|
|
|
fn rest(inner: &Mutex<Inner>, method: &str, path: &str, body: &[u8]) -> (u16, Value) {
|
|
let mut inner = inner.lock().unwrap();
|
|
inner.calls.push((method.to_string(), path.to_string()));
|
|
match (method, path) {
|
|
("GET", "/api/v4/users/me") if inner.me_status == 200 => {
|
|
(200, json!({"id": BOT, "username": BOT_NAME}))
|
|
}
|
|
("GET", "/api/v4/users/me") => (
|
|
inner.me_status,
|
|
json!({"id": "api.context.session_expired.app_error"}),
|
|
),
|
|
("POST", "/api/v4/channels/direct") => {
|
|
let users: Vec<String> = serde_json::from_slice(body).unwrap();
|
|
let channel = if users.iter().any(|u| u == KYLE) {
|
|
DM
|
|
} else {
|
|
EVE_DM
|
|
};
|
|
(201, json!({"id": channel, "type": "D"}))
|
|
}
|
|
("POST", "/api/v4/posts") => {
|
|
let p: Value = serde_json::from_slice(body).unwrap();
|
|
let n = u32::try_from(inner.posts.len()).unwrap();
|
|
let (channel, root, message) = (
|
|
p["channel_id"].as_str().unwrap(),
|
|
p["root_id"].as_str().unwrap(),
|
|
p["message"].as_str().unwrap(),
|
|
);
|
|
inner
|
|
.posts
|
|
.push((channel.to_string(), root.to_string(), message.to_string()));
|
|
(201, post(&id('x', n), BOT, channel, root, message, 1))
|
|
}
|
|
("GET", p) if p.contains("/posts?since=") => {
|
|
let channel = p
|
|
.trim_start_matches("/api/v4/channels/")
|
|
.split('/')
|
|
.next()
|
|
.unwrap_or("");
|
|
(
|
|
200,
|
|
inner
|
|
.since
|
|
.get(channel)
|
|
.cloned()
|
|
.unwrap_or(json!({"order": [], "posts": {}})),
|
|
)
|
|
}
|
|
_ => (404, json!({"message": "not found"})),
|
|
}
|
|
}
|