/build/source/nativelink-redis-tester/src/dynamic_fake_redis.rs

Source
// Copyright 2026 The NativeLink Authors. All rights reserved.
//
// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    See LICENSE file for details
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use core::fmt;
use std::collections::HashMap;
use std::collections::hash_map::Entry;
use std::sync::{Arc, Mutex};

use nativelink_util::background_spawn;
use redis::Value;
use redis_protocol::resp2::decode::decode;
use redis_protocol::resp2::types::{OwnedFrame, Resp2Frame};
use tokio::net::TcpListener;
use tokio::sync::oneshot::{self, Sender};
use tracing::{debug, info, trace};

use crate::fake_redis::{arg_as_string, fake_redis_internal};

pub trait SubscriptionManagerNotify {
    fn notify_for_test(&self, value: String);
}

#[derive(Clone)]
pub struct FakeRedisBackend<S: SubscriptionManagerNotify> {
    /// Contains a list of all of the Redis keys -> fields.
    pub table: Arc<Mutex<HashMap<String, HashMap<String, Value>>>>,
    /// TTL (seconds) attached to each key via `EXPIRE`, so tests can
    /// assert a key was given a bounded lifetime.
    pub expiries: Arc<Mutex<HashMap<String, i64>>>,
    subscription_manager: Arc<Mutex<Option<Arc<S>>>>,
}

impl<S: SubscriptionManagerNotify + Send + 'static + Sync> Default for FakeRedisBackend<S> {
    fn default() -> Self {
        Self::new()
    }
}

impl<S: SubscriptionManagerNotify> fmt::Debug for FakeRedisBackend<S> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("FakeRedisBackend").finish()
    }
}

const FAKE_SCRIPT_SHA: &str = "5148c724ce419ea27d1971dcb61c111dbbc6b63e";

impl<S: SubscriptionManagerNotify + Send + 'static + Sync> FakeRedisBackend<S> {
    pub fn new() -> Self {
        Self {
            table: Arc::new(Mutex::new(HashMap::new())),
            expiries: Arc::new(Mutex::new(HashMap::new())),
            subscription_manager: Arc::new(Mutex::new(None)),
        }
    }

    pub fn set_subscription_manager(&self, subscription_manager: Arc<S>) {
        self.subscription_manager
            .lock()
            .unwrap()
            .replace(subscription_manager);
    }

    async fn dynamic_fake_redis(self, listener: TcpListener, listener_ready_tx: Sender<()>) {
        let inner4 = move |buf: &[u8]| -> String {
            let mut output = String::new();
            let mut buf_index = 0;
            loop {
                let frame = match decode(&buf[buf_index..]).unwrap() {
                    Some((frame, amt)) => {
                        buf_index += amt;
                        frame
                    }
                    None => {
                        panic!("No frame!");
                    }
                };
                let (cmd, args) = {
                    if let OwnedFrame::Array(a) = frame {
                        if let OwnedFrame::BulkString(s) = a.first().unwrap() {
                            let args: Vec<_> = a[1..].to_vec();
                            (str::from_utf8(s).unwrap().to_string(), args)
                        } else {
                            panic!("Array not starting with cmd: {a:?}");
                        }
                    } else {
                        panic!("Non array cmd: {frame:?}");
                    }
                };

                let ret: Value = match cmd.as_str() {
                    "HELLO" => Value::Map(4vec!4[(
                        Value::SimpleString("server".into()),
                        Value::SimpleString("redis".into()),
                    )]),
                    "CLIENT" => {
                        // We can safely ignore these, as it's just setting the library name/version
                        Value::Int(0)
                    }
                    "SCRIPT" => {
                        assert_eq!(args[0], OwnedFrame::BulkString(b"LOAD".to_vec()));

                        let OwnedFrame::BulkString(ref _script) = args[1] else {
                            panic!("Script should be a bulkstring: {args:?}");
                        };
                        Value::SimpleString(FAKE_SCRIPT_SHA.to_string())
                    }

                    "PSUBSCRIBE" => {
                        // This does nothing at the moment, maybe we need to implement it later.
                        Value::Int(0)
                    }

                    "PUBLISH" => {
                        if let Some(subscription_manager) =
                            self.subscription_manager.lock().unwrap().as_ref()
                        {
                            subscription_manager.notify_for_test(
                                str::from_utf8(args[1].as_bytes().expect("Notification not bytes"))
                                    .expect("Notification not UTF-8")
                                    .into(),
                            );
                            Value::Int(1)
                        } else {
                            Value::Int(0)
                        }
                    }

                    "FT.AGGREGATE" => {
                        // The query is either "*" (match all) or @field:{ value }.
                        let OwnedFrame::BulkString(ref raw_query) = args[1] else {
                            panic!("Aggregate query should be a string: {args:?}");
                        };
                        let query = str::from_utf8(raw_query).unwrap();
                        // The real ft_aggregate caller now passes an explicit
                        // `TIMEOUT <ms>` clause before `LOAD`. Tolerate both
                        // shapes here so this fake doesn't break older callers
                        // and the LOAD-args check still validates the bit we
                        // actually care about.
                        let load_offset = if matches!(args.get(2), Some(OwnedFrame::BulkString(b)) if b == b"TIMEOUT")
                        {
                            // Skip "TIMEOUT" and its millisecond argument.
                            4
                        } else {
                            2
                        };
                        // Lazy implementation making assumptions.
                        assert_eq!(
                            args[load_offset..load_offset + 4],
                            vec![
                                OwnedFrame::BulkString(b"LOAD".to_vec()),
                                OwnedFrame::BulkString(b"2".to_vec()),
                                OwnedFrame::BulkString(b"data".to_vec()),
                                OwnedFrame::BulkString(b"version".to_vec())
                            ]
                        );
                        let mut results = vec![Value::Int(0)];

                        if query == "*" {
                            // Wildcard query - return all records that have both data and version fields.
                            // Some entries (e.g., from HSET) may not have version field.
                            for fields in self.table.lock().unwrap().values() {
                                if let (Some(data), Some(version)) =
                                    (fields.get("data"), fields.get("version"))
                                {
                                    results.push(Value::Array(vec![
                                        Value::BulkString(b"data".to_vec()),
                                        data.clone(),
                                        Value::BulkString(b"version".to_vec()),
                                        version.clone(),
                                    ]));
                                }
                            }
                        } else {
                            // Field-specific query: @field:{ value }
                            assert_eq!(&query[..1], "@");
                            let mut parts = query[1..].split(':');
                            let field = parts.next().expect("No field name");
                            let value = parts.next().expect("No value");
                            let value = value
                                .strip_prefix("{ ")
                                .and_then(|s| s.strip_suffix(" }"))
                                .unwrap_or(value);
                            for fields in self.table.lock().unwrap().values()23 {
                                if let Some(key_value19) = fields.get(field)
                                    && *key_value == Value::BulkString(value.as_bytes().to_vec())
                                {
                                    results.push(Value::Array(vec![
                                        Value::BulkString(b"data".to_vec()),
                                        fields.get("data").expect("No data field").clone(),
                                        Value::BulkString(b"version".to_vec()),
                                        fields.get("version").expect("No version field").clone(),
                                    ]));
                                }
                            }
                        }

                        results[0] =
                            Value::Int(i64::try_from(results.len() - 1).unwrap_or(i64::MAX));
                        Value::Array(vec![
                            Value::Array(results),
                            Value::Int(0), // Means no more items in cursor.
                        ])
                    }

                    "EVALSHA" => {
                        assert_eq!(
                            args[0],
                            OwnedFrame::BulkString(FAKE_SCRIPT_SHA.as_bytes().to_vec())
                        );
                        assert_eq!(args[1], OwnedFrame::BulkString(b"1".to_vec()));
                        let mut value: HashMap<_, Value> = HashMap::new();
                        value.insert(
                            "data".into(),
                            Value::BulkString(args[5].as_bytes().unwrap().to_vec()),
                        );
                        for pair in args[6..]17.chunks17(2) {
                            value.insert(
                                str::from_utf8(pair[0].as_bytes().expect("Field name not bytes"))
                                    .expect("Unable to parse field name as string")
                                    .into(),
                                Value::BulkString(pair[1].as_bytes().unwrap().to_vec()),
                            );
                        }
                        let mut ret: Option<Value> = None;
                        let key: String =
                            str::from_utf8(args[2].as_bytes().expect("Key not bytes"))
                                .expect("Key cannot be parsed as string")
                                .into();
                        let expected_existing_version: i64 =
                            str::from_utf8(args[3].as_bytes().unwrap())
                                .unwrap()
                                .parse()
                                .expect("Unable to parse existing version field");
                        let expiry: i64 = str::from_utf8(args[4].as_bytes().unwrap())
                            .unwrap()
                            .parse()
                            .expect("Unable to parse expiry field");
                        trace!(
                            key,
                            expected_existing_version,
                            expiry,
                            ?value,
                            "Want to insert with EVALSHA"
                        );
                        let version = match self.table.lock().unwrap().entry(key.clone()) {
                            Entry::Occupied(mut occupied_entry) => {
                                let version = occupied_entry
                                    .get()
                                    .get("version")
                                    .expect("No version field");
                                let Value::BulkString(version_bytes) = version else {
                                    panic!("Non-bulkstring version: {version:?}");
                                };
                                let version_int: i64 = str::from_utf8(version_bytes)
                                    .expect("Version field not valid string")
                                    .parse()
                                    .expect("Unable to parse version field");
                                if version_int == expected_existing_version {
                                    let new_version = version_int + 1;
                                    debug!(%key, %new_version, "Version update");
                                    value.insert(
                                        "version".into(),
                                        Value::BulkString(
                                            format!("{new_version}").as_bytes().to_vec(),
                                        ),
                                    );
                                    occupied_entry.insert(value);
                                    new_version
                                } else {
                                    // Version mismatch.
                                    debug!(%key, %version_int, %expected_existing_version, "Version mismatch");
                                    ret = Some(Value::Array(vec![
                                        Value::Int(0),
                                        Value::Int(version_int),
                                    ]));
                                    -1
                                }
                            }
                            Entry::Vacant(vacant_entry) => {
                                if expected_existing_version != 0 {
                                    // Version mismatch.
                                    debug!(%key, %expected_existing_version, "Version mismatch, expected zero");
                                    ret = Some(Value::Array(vec![Value::Int(0), Value::Int(0)]));
                                    -1
                                } else {
                                    debug!(%key, "Version insert");
                                    value
                                        .insert("version".into(), Value::BulkString(b"1".to_vec()));
                                    vacant_entry.insert_entry(value);
                                    1
                                }
                            }
                        };
                        if let Some(r4) = ret {
                            r
                        } else {
                            Value::Array(vec![Value::Int(1), Value::Int(version)])
                        }
                    }

                    "HMSET" => {
                        let mut values = HashMap::new();
                        assert_eq!(
                            (args.len() - 1).rem_euclid(2),
                            0,
                            "Non-even args for hmset: {args:?}"
                        );
                        let chunks = args[1..].chunks_exact(2);
                        for chunk in chunks {
                            let [key, value] = chunk else {
                                panic!("Uneven hmset args");
                            };
                            let key_name: String =
                                str::from_utf8(key.as_bytes().expect("Key argument is not bytes"))
                                    .expect("Unable to parse key as string")
                                    .into();
                            values.insert(
                                key_name,
                                Value::BulkString(value.as_bytes().unwrap().to_vec()),
                            );
                        }
                        let key =
                            str::from_utf8(args[0].as_bytes().expect("Key argument is not bytes"))
                                .expect("Unable to parse key as string")
                                .into();
                        debug!(%key, ?values, "Inserting with HMSET");
                        self.table.lock().unwrap().insert(key, values);
                        Value::Okay
                    }

                    "EXPIRE" => {
                        // `EXPIRE key seconds`: record the TTL; return 1
                        // if the key existed, else 0 (as real Redis does).
                        let key_name =
                            str::from_utf8(args[0].as_bytes().expect("Key argument is not bytes"))
                                .expect("Unable to parse key name")
                                .to_string();
                        let seconds = str::from_utf8(
                            args[1].as_bytes().expect("EXPIRE seconds is not bytes"),
                        )
                        .expect("EXPIRE seconds is not utf8")
                        .parse::<i64>()
                        .expect("EXPIRE seconds is not an integer");
                        let exists = self.table.lock().unwrap().contains_key(&key_name);
                        if exists {
                            self.expiries.lock().unwrap().insert(key_name, seconds);
                            Value::Int(1)
                        } else {
                            Value::Int(0)
                        }
                    }

                    "HMGET" => {
                        let key_name =
                            str::from_utf8(args[0].as_bytes().expect("Key argument is not bytes"))
                                .expect("Unable to parse key name");

                        if let Some(fields48) = self.table.lock().unwrap().get(key_name) {
                            trace!(%key_name, keys = ?fields.keys(), "Getting keys with HMGET, some keys");
                            let mut result = vec![];
                            for key in &args[1..]48 {
                                let field_name = str::from_utf8(
                                    key.as_bytes().expect("Field argument is not bytes"),
                                )
                                .expect("Unable to parse requested field");
                                if let Some(value94) = fields.get(field_name) {
                                    result.push(value.clone());
                                } else {
                                    debug!(%key_name, %field_name, "Missing field");
                                    result.push(Value::Nil);
                                }
                            }
                            Value::Array(result)
                        } else {
                            trace!(%key_name, "Getting keys with HMGET, empty");
                            let null_count = i64::try_from(args.len() - 1).unwrap();
                            Value::Array(vec![Value::Nil, Value::Int(null_count)])
                        }
                    }
                    actual => {
                        panic!("Mock command not implemented! {actual:?}");
                    }
                };

                arg_as_string(&mut output, ret);
                if buf_index == buf.len() {
                    break;
                }
            }
            output
        };
        fake_redis_internal(listener, listener_ready_tx, vec![inner]).await;
    }

    pub async fn run(self) -> u16 {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        info!("Using port {port}");

        let (listener_ready_tx, listener_ready_rx) = oneshot::channel::<()>();

        background_spawn!("listener", async move {
            self.dynamic_fake_redis(listener, listener_ready_tx).await;
        });

        listener_ready_rx
            .await
            .expect("Expected successful listener boot");

        port
    }
}

Coverage Report

Created: 2026-07-21 15:28

Line	Count	Source
1		// Copyright 2026 The NativeLink Authors. 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;
16		use std::collections::HashMap;
17		use std::collections::hash_map::Entry;
18		use std::sync::{Arc, Mutex};
19
20		use nativelink_util::background_spawn;
21		use redis::Value;
22		use redis_protocol::resp2::decode::decode;
23		use redis_protocol::resp2::types::{OwnedFrame, Resp2Frame};
24		use tokio::net::TcpListener;
25		use tokio::sync::oneshot::{self, Sender};
26		use tracing::{debug, info, trace};
27
28		use crate::fake_redis::{arg_as_string, fake_redis_internal};
29
30		pub trait SubscriptionManagerNotify {
31		fn notify_for_test(&self, value: String);
32		}
33
34		#[derive(Clone)]
35		pub struct FakeRedisBackend<S: SubscriptionManagerNotify> {
36		/// Contains a list of all of the Redis keys -> fields.
37		pub table: Arc<Mutex<HashMap<String, HashMap<String, Value>>>>,
38		/// TTL (seconds) attached to each key via `EXPIRE`, so tests can
39		/// assert a key was given a bounded lifetime.
40		pub expiries: Arc<Mutex<HashMap<String, i64>>>,
41		subscription_manager: Arc<Mutex<Option<Arc<S>>>>,
42		}
43
44		impl<S: SubscriptionManagerNotify + Send + 'static + Sync> Default for FakeRedisBackend<S> {
45	0	fn default() -> Self {
46	0	Self::new()
47	0	}
48		}
49
50		impl<S: SubscriptionManagerNotify> fmt::Debug for FakeRedisBackend<S> {
51	0	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
52	0	f.debug_struct("FakeRedisBackend").finish()
53	0	}
54		}
55
56		const FAKE_SCRIPT_SHA: &str = "5148c724ce419ea27d1971dcb61c111dbbc6b63e";
57
58		impl<S: SubscriptionManagerNotify + Send + 'static + Sync> FakeRedisBackend<S> {
59	4	pub fn new() -> Self {
60	4	Self {
61	4	table: Arc::new(Mutex::new(HashMap::new())),
62	4	expiries: Arc::new(Mutex::new(HashMap::new())),
63	4	subscription_manager: Arc::new(Mutex::new(None)),
64	4	}
65	4	}
66
67	3	pub fn set_subscription_manager(&self, subscription_manager: Arc<S>) {
68	3	self.subscription_manager
69	3	.lock()
70	3	.unwrap()
71	3	.replace(subscription_manager);
72	3	}
73
74	4	async fn dynamic_fake_redis(self, listener: TcpListener, listener_ready_tx: Sender<()>) {
75	169	let inner4 = move \|buf: &[u8]\| -> String {
76	169	let mut output = String::new();
77	169	let mut buf_index = 0;
78		loop {
79	183	let frame = match decode(&buf[buf_index..]).unwrap() {
80	183	Some((frame, amt)) => {
81	183	buf_index += amt;
82	183	frame
83		}
84		None => {
85	0	panic!("No frame!");
86		}
87		};
88	183	let (cmd, args) = {
89	183	if let OwnedFrame::Array(a) = frame {
90	183	if let OwnedFrame::BulkString(s) = a.first().unwrap() {
91	183	let args: Vec<_> = a[1..].to_vec();
92	183	(str::from_utf8(s).unwrap().to_string(), args)
93		} else {
94	0	panic!("Array not starting with cmd: {a:?}");
95		}
96		} else {
97	0	panic!("Non array cmd: {frame:?}");
98		}
99		};
100
101	183	let ret: Value = match cmd.as_str() {
102	183	"HELLO" => Value::Map(4 vec!4 [(
103	4	Value::SimpleString("server".into()),
104	4	Value::SimpleString("redis".into()),
105	4	)]),
106	179	"CLIENT" => {
107		// We can safely ignore these, as it's just setting the library name/version
108	8	Value::Int(0)
109		}
110	171	"SCRIPT" => {
111	4	assert_eq!(args[0], OwnedFrame::BulkString(b"LOAD".to_vec()));
112
113	4	let OwnedFrame::BulkString(ref _script) = args[1] else {
114	0	panic!("Script should be a bulkstring: {args:?}");
115		};
116	4	Value::SimpleString(FAKE_SCRIPT_SHA.to_string())
117		}
118
119	167	"PSUBSCRIBE" => {
120		// This does nothing at the moment, maybe we need to implement it later.
121	8	Value::Int(0)
122		}
123
124	159	"PUBLISH" => {
125	16	if let Some(subscription_manager) =
126	17	self.subscription_manager.lock().unwrap().as_ref()
127		{
128	16	subscription_manager.notify_for_test(
129	16	str::from_utf8(args[1].as_bytes().expect("Notification not bytes"))
130	16	.expect("Notification not UTF-8")
131	16	.into(),
132	16	);
133	16	Value::Int(1)
134		} else {
135	1	Value::Int(0)
136		}
137		}
138
139	142	"FT.AGGREGATE" => {
140		// The query is either "*" (match all) or @field:{ value }.
141	23	let OwnedFrame::BulkString(ref raw_query) = args[1] else {
142	0	panic!("Aggregate query should be a string: {args:?}");
143		};
144	23	let query = str::from_utf8(raw_query).unwrap();
145		// The real ft_aggregate caller now passes an explicit
146		// `TIMEOUT <ms>` clause before `LOAD`. Tolerate both
147		// shapes here so this fake doesn't break older callers
148		// and the LOAD-args check still validates the bit we
149		// actually care about.
150	23	let load_offset = if matches!(args.get(2), Some(OwnedFrame::BulkString(b)) if b == b"TIMEOUT")
151		{
152		// Skip "TIMEOUT" and its millisecond argument.
153	23	4
154		} else {
155	0	2
156		};
157		// Lazy implementation making assumptions.
158	23	assert_eq!(
159	23	args[load_offset..load_offset + 4],
160	23	vec![
161	23	OwnedFrame::BulkString(b"LOAD".to_vec()),
162	23	OwnedFrame::BulkString(b"2".to_vec()),
163	23	OwnedFrame::BulkString(b"data".to_vec()),
164	23	OwnedFrame::BulkString(b"version".to_vec())
165		]
166		);
167	23	let mut results = vec![Value::Int(0)];
168
169	23	if query == "*" {
170		// Wildcard query - return all records that have both data and version fields.
171		// Some entries (e.g., from HSET) may not have version field.
172	0	for fields in self.table.lock().unwrap().values() {
173	0	if let (Some(data), Some(version)) =
174	0	(fields.get("data"), fields.get("version"))
175	0	{
176	0	results.push(Value::Array(vec![
177	0	Value::BulkString(b"data".to_vec()),
178	0	data.clone(),
179	0	Value::BulkString(b"version".to_vec()),
180	0	version.clone(),
181	0	]));
182	0	}
183		}
184		} else {
185		// Field-specific query: @field:{ value }
186	23	assert_eq!(&query[..1], "@");
187	23	let mut parts = query[1..].split(':');
188	23	let field = parts.next().expect("No field name");
189	23	let value = parts.next().expect("No value");
190	23	let value = value
191	23	.strip_prefix("{ ")
192	23	.and_then(\|s\| s.strip_suffix(" }"))
193	23	.unwrap_or(value);
194	56	for fields in self.table.lock().unwrap().values()23 {
195	56	if let Some( key_value19 ) = fields.get(field)
196	19	&& *key_value == Value::BulkString(value.as_bytes().to_vec())
197	10	{
198	10	results.push(Value::Array(vec![
199	10	Value::BulkString(b"data".to_vec()),
200	10	fields.get("data").expect("No data field").clone(),
201	10	Value::BulkString(b"version".to_vec()),
202	10	fields.get("version").expect("No version field").clone(),
203	10	]));
204	46	}
205		}
206		}
207
208	23	results[0] =
209	23	Value::Int(i64::try_from(results.len() - 1).unwrap_or(i64::MAX));
210	23	Value::Array(vec![
211	23	Value::Array(results),
212	23	Value::Int(0), // Means no more items in cursor.
213	23	])
214		}
215
216	119	"EVALSHA" => {
217	17	assert_eq!(
218	17	args[0],
219	17	OwnedFrame::BulkString(FAKE_SCRIPT_SHA.as_bytes().to_vec())
220		);
221	17	assert_eq!(args[1], OwnedFrame::BulkString(b"1".to_vec()));
222	17	let mut value: HashMap<_, Value> = HashMap::new();
223	17	value.insert(
224	17	"data".into(),
225	17	Value::BulkString(args[5].as_bytes().unwrap().to_vec()),
226		);
227	51	for pair in args[6..]17 . chunks17 (2) {
228	51	value.insert(
229	51	str::from_utf8(pair[0].as_bytes().expect("Field name not bytes"))
230	51	.expect("Unable to parse field name as string")
231	51	.into(),
232	51	Value::BulkString(pair[1].as_bytes().unwrap().to_vec()),
233	51	);
234	51	}
235	17	let mut ret: Option<Value> = None;
236	17	let key: String =
237	17	str::from_utf8(args[2].as_bytes().expect("Key not bytes"))
238	17	.expect("Key cannot be parsed as string")
239	17	.into();
240	17	let expected_existing_version: i64 =
241	17	str::from_utf8(args[3].as_bytes().unwrap())
242	17	.unwrap()
243	17	.parse()
244	17	.expect("Unable to parse existing version field");
245	17	let expiry: i64 = str::from_utf8(args[4].as_bytes().unwrap())
246	17	.unwrap()
247	17	.parse()
248	17	.expect("Unable to parse expiry field");
249	17	trace!(
250		key,
251		expected_existing_version,
252		expiry,
253		?value,
254		"Want to insert with EVALSHA"
255		);
256	17	let version = match self.table.lock().unwrap().entry(key.clone()) {
257	13	Entry::Occupied(mut occupied_entry) => {
258	13	let version = occupied_entry
259	13	.get()
260	13	.get("version")
261	13	.expect("No version field");
262	13	let Value::BulkString(version_bytes) = version else {
263	0	panic!("Non-bulkstring version: {version:?}");
264		};
265	13	let version_int: i64 = str::from_utf8(version_bytes)
266	13	.expect("Version field not valid string")
267	13	.parse()
268	13	.expect("Unable to parse version field");
269	13	if version_int == expected_existing_version {
270	9	let new_version = version_int + 1;
271	9	debug!(%key, %new_version, "Version update");
272	9	value.insert(
273	9	"version".into(),
274	9	Value::BulkString(
275	9	format!("{new_version}").as_bytes().to_vec(),
276	9	),
277		);
278	9	occupied_entry.insert(value);
279	9	new_version
280		} else {
281		// Version mismatch.
282	4	debug!(%key, %version_int, %expected_existing_version, "Version mismatch");
283	4	ret = Some(Value::Array(vec![
284	4	Value::Int(0),
285	4	Value::Int(version_int),
286	4	]));
287	4	-1
288		}
289		}
290	4	Entry::Vacant(vacant_entry) => {
291	4	if expected_existing_version != 0 {
292		// Version mismatch.
293	0	debug!(%key, %expected_existing_version, "Version mismatch, expected zero");
294	0	ret = Some(Value::Array(vec![Value::Int(0), Value::Int(0)]));
295	0	-1
296		} else {
297	4	debug!(%key, "Version insert");
298	4	value
299	4	.insert("version".into(), Value::BulkString(b"1".to_vec()));
300	4	vacant_entry.insert_entry(value);
301	4	1
302		}
303		}
304		};
305	17	if let Some( r4 ) = ret {
306	4	r
307		} else {
308	13	Value::Array(vec![Value::Int(1), Value::Int(version)])
309		}
310		}
311
312	102	"HMSET" => {
313	4	let mut values = HashMap::new();
314	4	assert_eq!(
315	4	(args.len() - 1).rem_euclid(2),
316		0,
317		"Non-even args for hmset: {args:?}"
318		);
319	4	let chunks = args[1..].chunks_exact(2);
320	4	for chunk in chunks {
321	4	let [key, value] = chunk else {
322	0	panic!("Uneven hmset args");
323		};
324	4	let key_name: String =
325	4	str::from_utf8(key.as_bytes().expect("Key argument is not bytes"))
326	4	.expect("Unable to parse key as string")
327	4	.into();
328	4	values.insert(
329	4	key_name,
330	4	Value::BulkString(value.as_bytes().unwrap().to_vec()),
331		);
332		}
333	4	let key =
334	4	str::from_utf8(args[0].as_bytes().expect("Key argument is not bytes"))
335	4	.expect("Unable to parse key as string")
336	4	.into();
337	4	debug!(%key, ?values, "Inserting with HMSET");
338	4	self.table.lock().unwrap().insert(key, values);
339	4	Value::Okay
340		}
341
342	98	"EXPIRE" => {
343		// `EXPIRE key seconds`: record the TTL; return 1
344		// if the key existed, else 0 (as real Redis does).
345	4	let key_name =
346	4	str::from_utf8(args[0].as_bytes().expect("Key argument is not bytes"))
347	4	.expect("Unable to parse key name")
348	4	.to_string();
349	4	let seconds = str::from_utf8(
350	4	args[1].as_bytes().expect("EXPIRE seconds is not bytes"),
351	4	)
352	4	.expect("EXPIRE seconds is not utf8")
353	4	.parse::<i64>()
354	4	.expect("EXPIRE seconds is not an integer");
355	4	let exists = self.table.lock().unwrap().contains_key(&key_name);
356	4	if exists {
357	4	self.expiries.lock().unwrap().insert(key_name, seconds);
358	4	Value::Int(1)
359		} else {
360	0	Value::Int(0)
361		}
362		}
363
364	94	"HMGET" => {
365	94	let key_name =
366	94	str::from_utf8(args[0].as_bytes().expect("Key argument is not bytes"))
367	94	.expect("Unable to parse key name");
368
369	94	if let Some( fields48 ) = self.table.lock().unwrap().get(key_name) {
370	48	trace!(%key_name, keys = ?fields.keys(), "Getting keys with HMGET, some keys");
371	48	let mut result = vec![];
372	96	for key in &args[1..]48 {
373	96	let field_name = str::from_utf8(
374	96	key.as_bytes().expect("Field argument is not bytes"),
375		)
376	96	.expect("Unable to parse requested field");
377	96	if let Some( value94 ) = fields.get(field_name) {
378	94	result.push(value.clone());
379	94	} else {
380	2	debug!(%key_name, %field_name, "Missing field");
381	2	result.push(Value::Nil);
382		}
383		}
384	48	Value::Array(result)
385		} else {
386	46	trace!(%key_name, "Getting keys with HMGET, empty");
387	46	let null_count = i64::try_from(args.len() - 1).unwrap();
388	46	Value::Array(vec![Value::Nil, Value::Int(null_count)])
389		}
390		}
391	0	actual => {
392	0	panic!("Mock command not implemented! {actual:?}");
393		}
394		};
395
396	183	arg_as_string(&mut output, ret);
397	183	if buf_index == buf.len() {
398	169	break;
399	14	}
400		}
401	169	output
402	169	};
403	4	fake_redis_internal(listener, listener_ready_tx, vec![inner]).await;
404	0	}
405
406	4	pub async fn run(self) -> u16 {
407	4	let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
408	4	let port = listener.local_addr().unwrap().port();
409	4	info!("Using port {port}");
410
411	4	let (listener_ready_tx, listener_ready_rx) = oneshot::channel::<()>();
412
413	4	background_spawn!("listener", async move {
414	4	self.dynamic_fake_redis(listener, listener_ready_tx).await;
415	0	});
416
417	4	listener_ready_rx
418	4	.await
419	4	.expect("Expected successful listener boot");
420
421	4	port
422	4	}
423		}