// Copyright 2018 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

use failure::{err_msg, Error};
use fidl::endpoints::{RequestStream, ServerEnd};
use fidl_fuchsia_stash::{
    GetIteratorMarker, GetIteratorRequest, GetIteratorRequestStream, ListIteratorMarker,
    ListIteratorRequest, ListIteratorRequestStream, Value, MAX_KEY_SIZE, MAX_STRING_SIZE,
    ListItem, KeyValue
};
use fuchsia_async as fasync;
use fuchsia_syslog::{fx_log_err, fx_log_info};
use fuchsia_zircon_sys;
use futures::{TryFutureExt, TryStreamExt};
use parking_lot::Mutex;
use std::collections::HashMap;
use std::sync::Arc;

use crate::store;

// 16 byte overhead for vectors
// 16 byte overhead per string
// 1 byte used for the uint8 type enum
const LIST_PREFIX_CHUNK_SIZE: usize = (((fuchsia_zircon_sys::ZX_CHANNEL_MAX_MSG_BYTES as u64) - 16)
                                            / (MAX_KEY_SIZE + 16 + 1)) as usize;

// 16 byte overhead for vectors
// 16 byte overhead per string
const GET_PREFIX_CHUNK_SIZE: usize = (((fuchsia_zircon_sys::ZX_CHANNEL_MAX_MSG_BYTES as u64) - 16)
                                            / (MAX_KEY_SIZE + 16 + MAX_STRING_SIZE + 16)) as usize;

/// Accessors are used by clients to view and interact with the store. In this situation, that
/// means handling transactional logic and using the store_manager to perform store
/// lookups/manipulations.
pub struct Accessor {
    store_manager: Arc<Mutex<store::StoreManager>>,
    client_name: String,
    enable_bytes: bool,
    read_only: bool,
    fields_updated: Arc<Mutex<HashMap<store::Key, Option<Value>>>>,
}

impl Accessor {
    /// Create a new accessor for interacting with the store. The client_name field determines what
    /// namespace in the store will be used.
    pub fn new(
        store_manager: Arc<Mutex<store::StoreManager>>,
        enable_bytes: bool,
        read_only: bool,
        client_name: String,
    ) -> Accessor {
        Accessor {
            store_manager,
            enable_bytes,
            client_name,
            read_only,
            fields_updated: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    /// Retrieves a value from the store. If a value has been modified and commit() hasn't been
    /// called yet, the modified version will be returned.
    pub fn get_value(&self, key: &str) -> Result<Option<Value>, Error> {
        fx_log_info!("retrieving value for key: {}", key);
        let store_manager = self.store_manager.lock();

        if let Some(o_val) = self.fields_updated.lock().get(key) {
            // This accessor has an uncommited update for the field, so let's return that.
            match (self.enable_bytes, o_val) {
                (false, Some(Value::Bytesval(_))) => {
                    // Bytes are disabled. Return an error instead.
                    return Err(err_msg(
                        "client attempted to access bytes when the type is disabled",
                    ));
                }
                (_, Some(val)) => {
                    return Ok(Some(store::clone_value(val)?));
                }
                (_, None) => return Ok(None),
            }
        }

        let res = store_manager.get_value(&self.client_name, &key);

        match res {
            None => Ok(None),
            Some(Value::Bytesval(_)) if !self.enable_bytes => Err(err_msg(
                "client attempted to access bytes when the type is disabled",
            )),
            Some(val) => Ok(Some(store::clone_value(val)?)),
        }
    }

    /// Sets a value in the store. commit() must be called for this to take effect.
    pub fn set_value(&mut self, key: String, val: Value) -> Result<(), Error> {
        if self.read_only {
            return Err(err_msg(
                "client attempted to set a value with a read-only accessor",
            ));
        }
        if !self.enable_bytes {
            if let Value::Bytesval(_) = val {
                return Err(err_msg(
                    "client attempted to set bytes when the type is disabled",
                ));
            }
        }

        self.fields_updated.lock().insert(key, Some(val));

        Ok(())
    }

    /// Deletes a value from the store. commit() must be called for this to take effect.
    pub fn delete_value(&mut self, key: String) -> Result<(), Error> {
        if self.read_only {
            return Err(err_msg(
                "client attempted to delete a value with a read-only accessor",
            ));
        }

        self.fields_updated.lock().insert(key, None);
        Ok(())
    }

    /// Given a server endpoint, answers calls to get_next on the endpoint to return key/type pairs
    /// where the key contains the given prefix.
    pub fn list_prefix(
        &mut self,
        prefix: String,
        server_end: ServerEnd<ListIteratorMarker>,
    ) {
        let mut list_results = self
            .store_manager
            .lock()
            .list_prefix(&self.client_name, &prefix);

        // Merge the results with the pending updated fields
        let fields_updated = self.fields_updated.lock();
        if !fields_updated.is_empty() {
            // Delete any fields that have been marked for removal
            list_results.retain(|li| field_has_been_deleted(fields_updated.get(&li.key)));
            // Update any fields that have been updated/created
            for (key, o_val) in fields_updated.iter() {
                let mut key_was_added = false;
                for li in list_results.iter_mut() {
                    if let Some(v) = o_val {
                        if &li.key == key {
                            key_was_added = true;
                            li.type_ = store::value_to_type(&v);
                        }
                    }
                }
                // This is a new key, append it to the results.
                if !key_was_added {
                    if let Some(v) = o_val {
                        list_results.push(ListItem {
                            key: key.clone(),
                            type_: store::value_to_type(&v),
                        });
                    }
                }
            }
        }

        fasync::spawn(async move {
            let server_chan = fasync::Channel::from_channel(server_end.into_channel())?;
            let mut stream = ListIteratorRequestStream::from_channel(server_chan);
            while let Some(ListIteratorRequest::GetNext{ responder })
                = await!(stream.try_next())?
            {
                let split_at = list_results.len() - LIST_PREFIX_CHUNK_SIZE.min(list_results.len());
                let mut chunk = list_results.split_off(split_at);
                responder.send(&mut chunk.iter_mut())?;
            }
            Ok(())
        }.unwrap_or_else(|e: failure::Error| fx_log_err!("error running list prefix interface: {:?}", e)));
    }

    /// Given a server endpoint, answers calls to get_next on the endpoint to return key/value
    /// pairs where the key contains the given prefix.
    pub fn get_prefix(
        &mut self,
        prefix: String,
        server_end: ServerEnd<GetIteratorMarker>,
    ) -> Result<(), Error> {

        let mut get_results = self
            .store_manager
            .lock()
            .get_prefix(&self.client_name, &prefix)?;

        // Merge the results with the pending updated fields
        let fields_updated = self.fields_updated.lock();
        if !fields_updated.is_empty() {
            // Delete any fields that have been marked for removal
            get_results.retain(|kv| field_has_been_deleted(fields_updated.get(&kv.key)));
            // Update any fields that have been updated/created
            for (key, o_val) in fields_updated.iter() {
                let mut key_was_added = false;
                for kv in get_results.iter_mut() {
                    if let Some(v) = o_val {
                        if &kv.key == key {
                            key_was_added = true;
                            kv.val = store::clone_value(v)?;
                        }
                    }
                }
                // This is a new key, append it to the results.
                if !key_was_added {
                    if let Some(v) = o_val {
                        get_results.push(KeyValue {
                            key: key.clone(),
                            val: store::clone_value(v)?,
                        });
                    }
                }
            }
        }

        let enable_bytes = self.enable_bytes;

        fasync::spawn(async move {
            let server_chan = fasync::Channel::from_channel(server_end.into_channel())?;
            let mut stream = GetIteratorRequestStream::from_channel(server_chan);
            while let Some(GetIteratorRequest::GetNext{ responder })
                = await!(stream.try_next())?
            {
                let split_at = get_results.len() - GET_PREFIX_CHUNK_SIZE.min(get_results.len());
                let mut chunk = get_results.split_off(split_at);
                if !enable_bytes {
                    for item in chunk.iter() {
                        if let Value::Bytesval(_) = item.val {
                            Err(err_msg("client attempted to access bytes when the type is disabled"))?;
                        }
                    }
                }
                responder.send(&mut chunk.iter_mut())?;
            }
            Ok(())
        }.unwrap_or_else(|e: failure::Error| fx_log_err!("error running get prefix interface: {:?}", e)));
        Ok(())
    }

    /// Deletes all keys under a given prefix. commit() must be called for this to take effect.
    pub fn delete_prefix(&mut self, prefix: String) -> Result<(), Error> {
        if self.read_only {
            return Err(err_msg(
                "client attempted to delete a prefix with a read-only accessor",
            ));
        }

        let sm = self.store_manager.lock();
        let mut fields_updated = self.fields_updated.lock();

        let list_results = sm.list_prefix(&self.client_name, &prefix);

        for (key, o_val) in fields_updated.iter_mut() {
            if key.starts_with(&prefix) {
                *o_val = None;
            }
        }
        for li in list_results {
            fields_updated.insert(li.key, None);
        }
        Ok(())
    }

    /// Causes all state modifications to take effect and be written to disk atomically.
    pub fn commit(&mut self) -> Result<(), Error> {
        if self.read_only {
            return Err(err_msg(
                "client attempted to commit with a read-only accessor",
            ));
        }

        let mut store_manager = self.store_manager.lock();
        let mut fields_updated = self.fields_updated.lock();
        let mut old_values: HashMap<String, Option<Value>> = HashMap::new();

        // Iterate over each key updated. For each key, record the current value and then set or
        // delete it as appropriate. If an error is encountered, break out of the loop and use the
        // values we just recorded to undo the changes we made.

        let commit_result = (|| {
            for (key, o_val) in fields_updated.iter() {
                {
                    let o_old_value = store_manager.get_value(&self.client_name, key);
                    let old_value = match o_old_value {
                        Some(val_ref) => Some(store::clone_value(val_ref)?),
                        None => None,
                    };
                    old_values.insert(key.clone(), old_value);
                }

                match o_val {
                    Some(val) => store_manager.set_value(
                        &self.client_name,
                        key.clone(),
                        store::clone_value(val)?,
                    )?,
                    None => store_manager.delete_value(&self.client_name, key)?,
                }
            }
            Ok(())
        })();
        fields_updated.clear();
        if commit_result.is_ok() {
            return commit_result;
        }
        // There was a problem. Revert the changes.
        for (key, o_val) in old_values.iter() {
            match o_val {
                // TODO: on an error here, should we bail out immediately (current response),
                // or continue and attempt to revert as many changes as we can? Does a
                // partially-committed accessor mean that the store is now corrupted?
                Some(val) => {
                    store_manager.set_value(
                        &self.client_name,
                        key.clone(),
                        store::clone_value(val)?,
                    )?;
                }
                None => {
                    store_manager.delete_value(&self.client_name, key)?;
                }
            }
        }
        return commit_result;
    }
}

// A helper function for use in list_prefix and get_prefix
fn field_has_been_deleted<T>(f: Option<&Option<T>>) -> bool {
    match f {
        // The key has not been updated, keep it
        None => true,
        // The key has been marked for removal, drop it
        Some(None) => false,
        // The key has been updated to a new value, keep it
        Some(Some(_)) => true,
    }
}

#[cfg(test)]
mod tests {
    use crate::accessor::*;
    use crate::store;
    use fidl::client::QueryResponseFut;
    use fidl::endpoints::create_proxy;
    use fidl_fuchsia_stash::{ListItem, Value};
    use fuchsia_async as fasync;
    use parking_lot::Mutex;
    use std::sync::Arc;
    use tempfile::TempDir;

    fn get_tmp_store_manager(tmp_dir: &TempDir) -> store::StoreManager {
        store::StoreManager::new(tmp_dir.path().join("stash.store")).unwrap()
    }

    async fn drain_stash_iterator<T, F>(mut f: F) -> Vec<T>
    where
        F: FnMut() -> QueryResponseFut<Vec<T>>
    {
        let mut res = Vec::new();
        loop {
            let mut subset = await!(f()).unwrap();
            if subset.len() == 0 {
                break;
            }
            res.append(&mut subset);
        }
        res
    }

    #[test]
    fn test_get_value() {
        let test_client_name = "test_client".to_string();
        let test_key = "test_key".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let acc = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        sm.lock()
            .set_value(&test_client_name, test_key.clone(), Value::Boolval(true))
            .unwrap();

        let fetched_val = acc.get_value(&test_key).unwrap();
        assert_eq!(Some(Value::Boolval(true)), fetched_val);
        assert_eq!(0, acc.fields_updated.lock().len());
    }

    #[test]
    fn test_set_value() {
        let test_client_name = "test_client".to_string();
        let test_key = "test_key".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let mut acc = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        let test_val = Value::Boolval(true);
        let expected_val = Value::Boolval(true);

        acc.set_value(test_key.clone(), test_val).unwrap();
        assert_eq!(None, sm.lock().get_value(&test_client_name, &test_key));
        assert_eq!(1, acc.fields_updated.lock().len());
        assert_eq!(
            &Some(expected_val),
            acc.fields_updated.lock().get(&test_key).unwrap()
        );

        assert_eq!(None, sm.lock().get_value(&test_client_name, &test_key));

        acc.commit().unwrap();
        assert_eq!(
            &Value::Boolval(true),
            sm.lock().get_value(&test_client_name, &test_key).unwrap()
        );
        assert_eq!(0, acc.fields_updated.lock().len());
    }

    #[test]
    fn test_delete_value() {
        let test_client_name = "test_client".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let mut acc = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        sm.lock()
            .set_value(&test_client_name, "a".to_string(), Value::Boolval(true))
            .unwrap();

        assert_eq!(0, acc.fields_updated.lock().len());
        acc.delete_value("b".to_string()).unwrap();
        assert_eq!(1, acc.fields_updated.lock().len());

        acc.delete_value("a".to_string()).unwrap();
        assert_eq!(2, acc.fields_updated.lock().len());
        assert_eq!(&None, acc.fields_updated.lock().get("a").unwrap());

        acc.delete_value("a".to_string()).unwrap();
        assert_eq!(2, acc.fields_updated.lock().len());
        assert_eq!(&None, acc.fields_updated.lock().get("a").unwrap());

        assert_eq!(
            Some(&Value::Boolval(true)),
            sm.lock().get_value(&test_client_name, &"a".to_string())
        );

        acc.commit().unwrap();
        assert_eq!(
            None,
            sm.lock().get_value(&test_client_name, &"a".to_string())
        );
        assert_eq!(0, acc.fields_updated.lock().len());
    }

    #[test]
    fn test_list_prefix() {
        let test_client_name = "test_client".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let mut acc = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        let mut executor = fasync::Executor::new().unwrap();

        for key in vec!["a", "a/a", "a/b", "a/a/b", "b", "b/c", "bbbbb"] {
            sm.lock()
                .set_value(&test_client_name, key.to_string(), Value::Boolval(true))
                .unwrap();
        }

        let mut run_test = |prefix: &str, mut expected: Vec<&str>| {
            let (list_iterator, server_end) = create_proxy().unwrap();
            acc.list_prefix(prefix.to_string(), server_end);

            let actual = executor.run_singlethreaded(drain_stash_iterator(|| list_iterator.get_next()));
            let mut actual : Vec<String> = actual.iter().map(|li| li.key.clone()).collect();

            expected.sort_unstable();
            actual.sort_unstable();
            assert_eq!(expected, actual);
        };

        run_test("", vec!["a", "a/a", "a/b", "a/a/b", "b", "b/c", "bbbbb"]);
        run_test("a", vec!["a", "a/a", "a/b", "a/a/b"]);
        run_test("a/a", vec!["a/a", "a/a/b"]);
        run_test("b", vec!["b", "b/c", "bbbbb"]);
        run_test("bb", vec!["bbbbb"]);
        run_test("c", vec![]);
    }

    #[test]
    fn test_list_prefix_paging() {
        let test_client_name = "test_client".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let mut acc = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        let mut executor = fasync::Executor::new().unwrap();

        for key in vec!["a", "a/a", "a/b", "a/a/b", "b", "b/c", "bbbbb"] {
            sm.lock()
                .set_value(&test_client_name, key.to_string(), Value::Boolval(true))
                .unwrap();
        }

        let mut run_test = |prefix: &str, mut expected: Vec<&str>| {
            let (list_iterator, server_end) = create_proxy().unwrap();
            acc.list_prefix(prefix.to_string(), server_end);

            let actual = executor.run_singlethreaded(drain_stash_iterator(|| list_iterator.get_next()));
            let mut actual : Vec<String> = actual.iter().map(|li| li.key.clone()).collect();

            expected.sort_unstable();
            actual.sort_unstable();
            assert_eq!(expected, actual);
        };

        run_test("", vec!["a", "a/a", "a/b", "a/a/b", "b", "b/c", "bbbbb"]);
        run_test("a", vec!["a", "a/a", "a/b", "a/a/b"]);
        run_test("a/a", vec!["a/a", "a/a/b"]);
        run_test("b", vec!["b", "b/c", "bbbbb"]);
        run_test("bb", vec!["bbbbb"]);
        run_test("c", vec![]);
    }

    #[test]
    fn test_get_prefix() {
        let test_client_name = "test_client".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let mut acc = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        let mut executor = fasync::Executor::new().unwrap();

        for (key, val) in vec![
            ("a", true),
            ("a/a", true),
            ("a/b", false),
            ("a/a/b", false),
            ("b", false),
            ("b/c", true),
            ("bbbbb", true),
        ] {
            sm.lock()
                .set_value(&test_client_name, key.to_string(), Value::Boolval(val))
                .unwrap();
        }

        let mut run_test = |prefix: &str, mut expected: Vec<(String, bool)>| {
            let (get_iterator, server_end) = create_proxy().unwrap();
            acc.get_prefix(prefix.to_string(), server_end).unwrap();

            let mut actual = Vec::new();
            loop {
                let subset = executor
                    .run_singlethreaded(get_iterator.get_next())
                    .unwrap();
                if subset.len() == 0 {
                    break;
                }
                for kv in subset.iter() {
                    let b = match kv.val {
                        Value::Boolval(b) => b,
                        _ => panic!("internal test error"),
                    };
                    actual.push((kv.key.clone(), b));
                }
            }
            expected.sort_unstable();
            actual.sort_unstable();
            assert_eq!(expected, actual);
        };

        run_test(
            "",
            vec![
                ("a".to_string(), true),
                ("a/a".to_string(), true),
                ("a/b".to_string(), false),
                ("a/a/b".to_string(), false),
                ("b".to_string(), false),
                ("b/c".to_string(), true),
                ("bbbbb".to_string(), true),
            ],
        );
        run_test(
            "a",
            vec![
                ("a".to_string(), true),
                ("a/a".to_string(), true),
                ("a/b".to_string(), false),
                ("a/a/b".to_string(), false),
            ],
        );
        run_test(
            "a/a",
            vec![("a/a".to_string(), true), ("a/a/b".to_string(), false)],
        );
        run_test(
            "b",
            vec![
                ("b".to_string(), false),
                ("b/c".to_string(), true),
                ("bbbbb".to_string(), true),
            ],
        );
        run_test("bb", vec![("bbbbb".to_string(), true)]);
        run_test("c", vec![]);
    }

    #[test]
    fn test_delete_prefix() {
        let test_client_name = "test_client".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let mut acc = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        let mut executor = fasync::Executor::new().unwrap();

        for key in vec!["a", "a/a", "a/b", "a/a/b", "b", "b/c", "bbbbb"] {
            sm.lock()
                .set_value(&test_client_name, key.to_string(), Value::Boolval(true))
                .unwrap();
        }

        acc.delete_prefix("a".to_string()).unwrap();
        assert_eq!(&None, acc.fields_updated.lock().get("a").unwrap());
        assert_eq!(&None, acc.fields_updated.lock().get("a/a").unwrap());
        assert_eq!(&None, acc.fields_updated.lock().get("a/b").unwrap());
        assert_eq!(&None, acc.fields_updated.lock().get("a/a/b").unwrap());
        assert_eq!(4, acc.fields_updated.lock().len());
        acc.commit().unwrap();

        let (list_iterator, server_end) = create_proxy().unwrap();
        acc.list_prefix("".to_string(), server_end);

        let mut actual = Vec::new();
        loop {
            let subset = executor
                .run_singlethreaded(list_iterator.get_next())
                .unwrap();
            if subset.len() == 0 {
                break;
            }
            for ListItem { key, type_: _ } in subset.iter() {
                actual.push(key.clone());
            }
        }
        actual.sort_unstable();
        assert_eq!(vec!["b", "b/c", "bbbbb"], actual);
    }

    #[test]
    fn test_commit_different_accessors() {
        let test_client_name = "test_client".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let mut acc1 = Accessor::new(sm.clone(), true, false, test_client_name.clone());
        let mut acc2 = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        {
            let mut sm = sm.lock();
            sm.set_value(&test_client_name, "a".to_string(), Value::Boolval(true))
                .unwrap();
            sm.set_value(&test_client_name, "b".to_string(), Value::Boolval(false))
                .unwrap();
        }

        acc1.get_value("a").unwrap();
        acc1.set_value("b".to_string(), Value::Boolval(true))
            .unwrap();

        acc1.commit().unwrap();

        assert_eq!(
            &Value::Boolval(true),
            sm.lock()
                .get_value(&test_client_name, &"b".to_string())
                .unwrap()
        );

        acc2.get_value("a").unwrap();
        acc2.delete_value("b".to_string()).unwrap();

        acc2.commit().unwrap();

        assert_eq!(
            None,
            sm.lock().get_value(&test_client_name, &"b".to_string())
        );

        // Confirm that the fields touched and fields updated maps have been cleared
        assert_eq!(0, acc1.fields_updated.lock().len());
        assert_eq!(0, acc2.fields_updated.lock().len());
    }

    #[test]
    fn test_commit_reuse_accessor() {
        let test_client_name = "test_client".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let mut acc1 = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        {
            let mut sm = sm.lock();
            sm.set_value(&test_client_name, "a".to_string(), Value::Boolval(true))
                .unwrap();
            sm.set_value(&test_client_name, "b".to_string(), Value::Boolval(false))
                .unwrap();
        }

        acc1.get_value("a").unwrap();
        acc1.set_value("b".to_string(), Value::Boolval(true))
            .unwrap();

        acc1.commit().unwrap();
        assert_eq!(0, acc1.fields_updated.lock().len());

        assert_eq!(
            &Value::Boolval(true),
            sm.lock()
                .get_value(&test_client_name, &"b".to_string())
                .unwrap()
        );

        acc1.get_value("a").unwrap();
        acc1.delete_value("b".to_string()).unwrap();

        acc1.commit().unwrap();
        assert_eq!(0, acc1.fields_updated.lock().len());

        assert_eq!(
            None,
            sm.lock().get_value(&test_client_name, &"b".to_string())
        );
    }

    #[test]
    fn test_delete_prefix_considers_current_commit() {
        let test_client_name = "test_client".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let mut acc = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        for key in vec!["a", "aa", "aaa"] {
            acc.set_value(key.to_owned(), Value::Boolval(true)).unwrap();
        }
        acc.commit().unwrap();
        for key in vec!["b", "bb", "bbb"] {
            acc.set_value(key.to_owned(), Value::Boolval(true)).unwrap();
        }

        acc.delete_prefix("".to_string()).unwrap();

        let mut keys : Vec<String> = acc.fields_updated.lock().keys().map(|s| s.clone()).collect();
        keys.sort_unstable();
        assert_eq!(vec!["a", "aa", "aaa", "b", "bb" ,"bbb"], keys);
        for key in vec!["a", "aa", "aaa", "b", "bb" ,"bbb"] {
            assert_eq!(Some(&None), acc.fields_updated.lock().get(key));
        }
    }

    #[test]
    fn test_list_prefix_considers_current_commit() {
        let mut executor = fasync::Executor::new().unwrap();

        let test_client_name = "test_client".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let mut acc = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        for key in vec!["a", "aa", "aaa"] {
            acc.set_value(key.to_owned(), Value::Boolval(true)).unwrap();
        }
        acc.commit().unwrap();
        for key in vec!["b", "bb", "bbb"] {
            acc.set_value(key.to_owned(), Value::Boolval(true)).unwrap();
        }

        let (list_iterator, server_end) = create_proxy().unwrap();
        acc.list_prefix("".to_string(), server_end);

        let res = executor.run_singlethreaded(drain_stash_iterator(|| list_iterator.get_next()));
        let mut res : Vec<String> = res.iter().map(|li| li.key.clone()).collect();

        res.sort_unstable();
        assert_eq!(vec!["a", "aa", "aaa", "b", "bb" ,"bbb"], res);
    }

    #[test]
    fn test_get_prefix_considers_current_commit() {
        let mut executor = fasync::Executor::new().unwrap();

        let test_client_name = "test_client".to_string();

        let tmp_dir = TempDir::new().unwrap();
        let sm = Arc::new(Mutex::new(get_tmp_store_manager(&tmp_dir)));
        let mut acc = Accessor::new(sm.clone(), true, false, test_client_name.clone());

        for key in vec!["a", "aa", "aaa"] {
            acc.set_value(key.to_owned(), Value::Boolval(true)).unwrap();
        }
        acc.commit().unwrap();
        for key in vec!["b", "bb", "bbb"] {
            acc.set_value(key.to_owned(), Value::Boolval(true)).unwrap();
        }

        let (get_iterator, server_end) = create_proxy().unwrap();
        acc.get_prefix("".to_string(), server_end).unwrap();

        let res = executor.run_singlethreaded(drain_stash_iterator(|| get_iterator.get_next()));
        let mut res : Vec<String> = res.iter().map(|kv| kv.key.clone()).collect();

        res.sort_unstable();
        assert_eq!(vec!["a", "aa", "aaa", "b", "bb" ,"bbb"], res);
    }
}