server.rs

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

//! server::run のテスト。実際に待ち受け、TCP でつないで確かめる。
//!
//! axum::serve は、ヘッダを受け取り終えるまでの時間に上限を掛けない。
//! ヘッダを送り終えない相手や、つないだまま何も送らない相手に、接続を持たれ続ける。
//! server::run は、その上限を REQUEST_TIMEOUT で掛ける。

use axum::Router;
use axum::routing::get;
use mon::state::AppState;
use mon::{ClientIp, Conf, database, relay, route, server, session};
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};


/// mon を 1 つ起こして、待ち受けているアドレスを返す
async fn start(timeout: Duration) -> SocketAddr {
    let conf = Arc::new(Conf {
        timeout,
        session_store: "none".to_owned(),
        ..Conf::default()
    });

    let database = database::connect("sqlite::memory:").await.unwrap();
    let store = session::store(&conf, &database);
    let routes = Router::new()
        .route("/", get(|| async { "ok" }))
        .route("/ip", get(where_from));

    let state = AppState {
        database,
        client: relay::client(&conf).unwrap(),
        session: store,
        conf: conf.clone(),
    };

    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let address = listener.local_addr().unwrap();
    let app = route::wrap(routes, state);

    tokio::spawn(async move {
        server::run(listener, app, &conf, std::future::pending()).await.unwrap();
    });

    address
}


async fn where_from(ClientIp(address): ClientIp) -> String {
    address.map(|value| value.to_string()).unwrap_or_default()
}


/// 相手が切るまで読む。3 秒たっても切られなければ失敗にする
async fn read_until_closed(stream: &mut TcpStream) -> String {
    let mut content = Vec::new();
    tokio::time::timeout(Duration::from_secs(3), stream.read_to_end(&mut content))
        .await
        .expect("3 秒たっても接続が切られません")
        .ok();

    String::from_utf8_lossy(&content).into_owned()
}


#[tokio::test]
async fn a_request_is_answered() {
    let address = start(Duration::from_secs(3)).await;
    let mut stream = TcpStream::connect(address).await.unwrap();
    stream
        .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();

    let content = read_until_closed(&mut stream).await;
    assert!(content.starts_with("HTTP/1.1 200"), "{content}");
    assert!(content.ends_with("ok"), "{content}");
}


/// TCP の相手が ClientIp に入る
#[tokio::test]
async fn the_peer_reaches_the_handler() {
    let address = start(Duration::from_secs(3)).await;
    let mut stream = TcpStream::connect(address).await.unwrap();
    stream
        .write_all(b"GET /ip HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();

    let content = read_until_closed(&mut stream).await;
    assert!(content.ends_with("127.0.0.1"), "{content}");
}


/// ヘッダを送り終えない相手は、上限の時間で切る
#[tokio::test]
async fn an_unfinished_header_is_cut_off() {
    let address = start(Duration::from_millis(300)).await;
    let mut stream = TcpStream::connect(address).await.unwrap();
    stream.write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n").await.unwrap();

    read_until_closed(&mut stream).await;
}


/// つないだまま何も送らない相手も、上限の時間で切る
#[tokio::test]
async fn a_silent_connection_is_cut_off() {
    let address = start(Duration::from_millis(300)).await;
    let mut stream = TcpStream::connect(address).await.unwrap();

    read_until_closed(&mut stream).await;
}