Coverage Report

Created: 2026-11-13 23:29

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/build/source/nativelink-redis-tester/src/fake_redis.rs
Line
Count
Source
1
// Copyright 2026 Trace Machina, Inc. All rights reserved.
2
//
3
// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//    See LICENSE file for details
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14
15
use core::fmt::Write;
16
use core::hash::BuildHasher;
17
use std::collections::HashMap;
18
19
use nativelink_util::background_spawn;
20
use redis::Value;
21
use redis_test::IntoRedisValue;
22
use tokio::io::{AsyncReadExt, AsyncWriteExt};
23
use tokio::net::TcpListener;
24
use tokio::sync::oneshot::{self, Sender};
25
use tracing::{error, info, warn};
26
27
152
fn cmd_as_string(cmd: &redis::Cmd) -> String {
28
152
    let raw = cmd.get_packed_command();
29
152
    String::from_utf8(raw).unwrap()
30
152
}
31
32
4.47M
pub(crate) fn arg_as_string(output: &mut String, arg: Value) {
33
4.47M
    match arg {
34
452
        Value::SimpleString(s) => {
35
452
            write!(output, "+{s}\r\n").unwrap();
36
452
        }
37
39.0k
        Value::Okay => {
38
39.0k
            write!(output, "+OK\r\n").unwrap();
39
39.0k
        }
40
2.11M
        Value::BulkString(s) => {
41
2.11M
            write!(
42
2.11M
                output,
43
2.11M
                "${}\r\n{}\r\n",
44
2.11M
                s.len(),
45
2.11M
                str::from_utf8(&s).unwrap()
46
2.11M
            )
47
2.11M
            .unwrap();
48
2.11M
        }
49
638k
        Value::Int(v) => {
50
638k
            write!(output, ":{v}\r\n").unwrap();
51
638k
        }
52
1.13M
        Value::Array(values) => {
53
1.13M
            write!(output, "*{}\r\n", values.len()).unwrap();
54
3.71M
            for value in 
values1.13M
{
55
3.71M
                arg_as_string(output, value);
56
3.71M
            }
57
        }
58
181
        Value::Map(values) => {
59
181
            write!(output, "%{}\r\n", values.len()).unwrap();
60
202
            for (key, value) in 
values181
{
61
202
                arg_as_string(output, key);
62
202
                arg_as_string(output, value);
63
202
            }
64
        }
65
541k
        Value::Nil => {
66
541k
            write!(output, "_\r\n").unwrap();
67
541k
        }
68
1
        Value::Boolean(value) => {
69
1
            if value {
70
1
                write!(output, "#t\r\n")
71
            } else {
72
0
                write!(output, "#f\r\n")
73
            }
74
1
            .unwrap();
75
        }
76
        _ => {
77
0
            panic!("No support for {arg:?}")
78
        }
79
    }
80
4.47M
}
81
82
126
fn args_as_string(args: Vec<Value>) -> String {
83
126
    let mut output = String::new();
84
291
    for arg in 
args126
{
85
291
        arg_as_string(&mut output, arg);
86
291
    }
87
126
    output
88
126
}
89
90
76
pub fn add_to_response<B: BuildHasher>(
91
76
    response: &mut HashMap<String, String, B>,
92
76
    cmd: &redis::Cmd,
93
76
    args: Vec<Value>,
94
76
) {
95
76
    add_to_response_raw(response, cmd, args_as_string(args));
96
76
}
97
98
77
pub fn add_to_response_raw<B: BuildHasher>(
99
77
    response: &mut HashMap<String, String, B>,
100
77
    cmd: &redis::Cmd,
101
77
    args: String,
102
77
) {
103
77
    response.insert(cmd_as_string(cmd), args);
104
77
}
105
106
25
fn setinfo(responses: &mut HashMap<String, String>) {
107
    // We do raw inserts of command here, because the library sends 3/4 commands in one go
108
    // They always start with HELLO, then optionally SELECT, so we use this to differentiate
109
25
    let hello = cmd_as_string(redis::cmd("HELLO").arg("3"));
110
25
    let setinfo = cmd_as_string(
111
25
        redis::cmd("CLIENT")
112
25
            .arg("SETINFO")
113
25
            .arg("LIB-NAME")
114
25
            .arg("redis-rs"),
115
    );
116
25
    responses.insert(
117
25
        [hello.clone(), setinfo.clone()].join(""),
118
25
        args_as_string(vec![
119
25
            Value::Map(vec![(
120
25
                Value::SimpleString("server".into()),
121
25
                Value::SimpleString("redis".into()),
122
25
            )]),
123
25
            Value::Okay,
124
25
            Value::Okay,
125
        ]),
126
    );
127
25
    responses.insert(
128
25
        [hello, cmd_as_string(redis::cmd("SELECT").arg(3)), setinfo].join(""),
129
25
        args_as_string(vec![
130
25
            Value::Map(vec![(
131
25
                Value::SimpleString("server".into()),
132
25
                Value::SimpleString("redis".into()),
133
25
            )]),
134
25
            Value::Okay,
135
25
            Value::Okay,
136
25
            Value::Okay,
137
        ]),
138
    );
139
25
}
140
141
18
pub fn add_lua_script<B: BuildHasher>(
142
18
    responses: &mut HashMap<String, String, B>,
143
18
    lua_script: &str,
144
18
    hash: &str,
145
18
) {
146
18
    add_to_response(
147
18
        responses,
148
18
        redis::cmd("SCRIPT").arg("LOAD").arg(lua_script),
149
18
        vec![hash.into_redis_value()],
150
    );
151
18
}
152
153
18
pub fn fake_redis_stream() -> HashMap<String, String> {
154
18
    let mut responses = HashMap::new();
155
18
    setinfo(&mut responses);
156
    // Does setinfo as well, so need to respond to all 3
157
18
    add_to_response(
158
18
        &mut responses,
159
18
        redis::cmd("SELECT").arg("3"),
160
18
        vec![Value::Okay, Value::Okay, Value::Okay],
161
    );
162
18
    responses
163
18
}
164
165
3
pub fn fake_redis_sentinel_master_stream() -> HashMap<String, String> {
166
3
    let mut response = fake_redis_stream();
167
3
    add_to_response(
168
3
        &mut response,
169
3
        &redis::cmd("ROLE"),
170
3
        vec![Value::Array(vec![
171
3
            "master".into_redis_value(),
172
3
            0.into_redis_value(),
173
3
            Value::Array(vec![]),
174
3
        ])],
175
    );
176
3
    response
177
3
}
178
179
7
pub fn fake_redis_sentinel_stream(master_name: &str, redis_port: u16) -> HashMap<String, String> {
180
7
    let mut response = HashMap::new();
181
7
    setinfo(&mut response);
182
183
    // Not a full "sentinel masters" response, but enough for redis-rs
184
7
    let resp: Vec<(Value, Value)> = vec![
185
7
        ("name".into_redis_value(), master_name.into_redis_value()),
186
7
        ("ip".into_redis_value(), "127.0.0.1".into_redis_value()),
187
7
        (
188
7
            "port".into_redis_value(),
189
7
            i64::from(redis_port).into_redis_value(),
190
7
        ),
191
7
        ("flags".into_redis_value(), "master".into_redis_value()),
192
    ];
193
194
7
    add_to_response(
195
7
        &mut response,
196
7
        redis::cmd("SENTINEL").arg("MASTERS"),
197
7
        vec![Value::Array(vec![Value::Map(resp)])],
198
    );
199
7
    response
200
7
}
201
202
29
pub(crate) async fn fake_redis_internal<H>(
203
29
    listener: TcpListener,
204
29
    listener_ready_tx: Sender<()>,
205
29
    handlers: Vec<H>,
206
29
) where
207
29
    H: Fn(&[u8]) -> String + Send + Clone + 'static + Sync,
208
29
{
209
29
    let mut handler_iter = handlers.iter().cloned().cycle();
210
29
    info!(
211
        "Waiting for connection on {}",
212
29
        listener.local_addr().unwrap()
213
    );
214
29
    listener_ready_tx
215
29
        .send(())
216
29
        .expect("Expected successful send");
217
    loop {
218
72
        let Ok((
mut stream43
, _)) = listener.accept().await else {
219
0
            error!("accept error");
220
0
            panic!("error");
221
        };
222
43
        info!("Accepted new connection");
223
43
        let local_handler = handler_iter.next().unwrap();
224
43
        background_spawn!("thread", async move {
225
            loop {
226
216k
                let mut buf = vec![0; 8192];
227
216k
                let 
res216k
= stream.read(&mut buf).await.
unwrap216k
();
228
216k
                if res != 0 {
229
128
                    let output = local_handler(&buf[..res]);
230
128
                    if !output.is_empty() {
231
124
                        stream.write_all(output.as_bytes()).await.unwrap();
232
4
                    }
233
216k
                }
234
            }
235
        });
236
    }
237
}
238
239
26
async fn fake_redis<B>(
240
26
    listener: TcpListener,
241
26
    listener_ready_tx: Sender<()>,
242
26
    all_responses: Vec<HashMap<String, String, B>>,
243
26
) where
244
26
    B: BuildHasher + Clone + Send + 'static + Sync,
245
26
{
246
26
    let funcs = all_responses
247
26
        .iter()
248
26
        .map(|responses| {
249
26
            info!("Responses are: {:?}", responses);
250
26
            let values = responses.clone();
251
99
            move |buf: &[u8]| -> String {
252
99
                let str_buf = String::from_utf8_lossy(buf).into_owned();
253
362
                for (key, value) in 
&values99
{
254
362
                    if str_buf.starts_with(key) {
255
95
                        info!("Responding to {}", str_buf.replace("\r\n", "\\r\\n"));
256
95
                        return value.clone();
257
267
                    }
258
                }
259
4
                warn!(
260
                    "Unknown command: {}",
261
4
                    str_buf.chars().take(1000).collect::<String>()
262
                );
263
4
                String::new()
264
99
            }
265
26
        })
266
26
        .collect();
267
26
    fake_redis_internal(listener, listener_ready_tx, funcs).await;
268
0
}
269
270
26
async fn make_fake_redis_with_multiple_responses<B: BuildHasher + Clone + Send + 'static + Sync>(
271
26
    responses: Vec<HashMap<String, String, B>>,
272
26
) -> u16 {
273
26
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
274
26
    let port = listener.local_addr().unwrap().port();
275
276
26
    let (listener_ready_tx, listener_ready_rx) = oneshot::channel::<()>();
277
278
26
    background_spawn!("listener", async move {
279
26
        fake_redis(listener, listener_ready_tx, responses).await;
280
0
    });
281
282
26
    listener_ready_rx
283
26
        .await
284
26
        .expect("Expected successful listener boot");
285
26
    info!(port, "Fake redis booted");
286
287
26
    port
288
26
}
289
290
26
pub async fn make_fake_redis_with_responses<B: BuildHasher + Clone + Send + 'static + Sync>(
291
26
    responses: HashMap<String, String, B>,
292
26
) -> u16 {
293
26
    make_fake_redis_with_multiple_responses(vec![responses]).await
294
26
}