// Copyright 2018 Zach Miller
//
// Licensed under the Apache License, Version 2.0 (http://www.apache.org/licenses/LICENSE-2.0)
// or the MIT License (https://opensource.org/licenses/MIT), at your option.
//
// This file may not be copied, modified, or distributed except according to those terms.

use bincode;
use serde::de::DeserializeOwned;
use serde::ser::Serialize;

use std::fs::{self, File};
use std::io::{BufReader, BufWriter};
use std::path::{Path, PathBuf};

use error::{Error, Result};
use storage::{CommittedCheckpoint, Storage, UncommittedCheckpoint};
use wrappers::GuardWrapper;

/// Provides checkpoint storage via the file system.
#[derive(Debug)]
pub struct FileStorage {
    committed_path: PathBuf,
    uncommitted_path: PathBuf,
}

impl FileStorage {
    /// Creates a `FileStorage` object around a new or existing file path and wraps the object in a
    /// [`GuardWrapper`](
    ///     https://docs.rs/checkpoint/0.1.5/checkpoint/wrappers/struct.GuardWrapper.html).
    ///
    /// If `path` already exists it must point to a directory which contains two sub-folders,
    /// "committed" and "uncommitted".
    pub fn open(path: &Path) -> Result<GuardWrapper<Self>> {
        let committed_path = path.join("committed");
        let uncommitted_path = path.join("uncommitted");

        if path.exists() {
            if !path.is_dir() {
                Err(Error::initialization(format!(
                    "{:?} is not a directory",
                    path
                )))?;
            }

            if !committed_path.exists() || !uncommitted_path.exists() {
                Err(Error::initialization(format!(
                    "{:?} does not contain checkpoint data",
                    path
                )))?;
            }
        } else {
            fs::create_dir_all(path).map_err(|_| {
                Error::io(format!("Could not create {:?} to store checkpoints", path))
            })?;

            fs::create_dir(&committed_path).map_err(|_| {
                Error::io(format!(
                    "Could not create {:?} to store committed checkpoints",
                    committed_path
                ))
            })?;

            fs::create_dir(&uncommitted_path).map_err(|_| {
                Error::io(format!(
                    "Could not create {:?} to store uncommitted checkpoints",
                    uncommitted_path
                ))
            })?;
        }

        Ok(GuardWrapper::wrap(Self {
            committed_path,
            uncommitted_path,
        })?)
    }
}

impl Storage for FileStorage {
    type Committed = Committed;
    type Uncommitted = Uncommitted;

    /// Creates a new checkpoint with the specified identifier.
    ///
    /// The checkpoint will not be saved permanently until it has been committed via the
    /// `commit_checkpoint` method.
    ///
    /// Note: The provided identifier must be a valid filename.
    fn create_checkpoint(&mut self, identifier: &str) -> Result<Self::Uncommitted> {
        let uncommitted_checkpoint_path = self.uncommitted_path.join(identifier);
        if uncommitted_checkpoint_path.exists() {
            fs::remove_dir_all(&uncommitted_checkpoint_path).map_err(|_| {
                Error::io(format!(
                    "Could not remove {:?} which contains uncommitted checkpoint data",
                    uncommitted_checkpoint_path
                ))
            })?;
        }

        fs::create_dir(&uncommitted_checkpoint_path).map_err(|_| {
            Error::io(format!(
                "Could not create {:?} to store uncommitted checkpoint data",
                uncommitted_checkpoint_path
            ))
        })?;

        Ok(Self::Uncommitted::new(
            identifier,
            &uncommitted_checkpoint_path,
        ))
    }

    fn commit_checkpoint(&mut self, mut uncommitted: Self::Uncommitted) -> Result<Self::Committed> {
        let identifier = uncommitted.identifier()?.to_string();
        let committed_checkpoint_path = self.committed_path.join(&identifier);

        fs::rename(&uncommitted.inner.path, &committed_checkpoint_path).map_err(|_| {
            Error::io(format!(
                "Could not create {:?} to store committed checkpoint data",
                committed_checkpoint_path
            ))
        })?;

        Ok(Self::Committed::new(
            &identifier,
            &committed_checkpoint_path,
        ))
    }

    fn load_checkpoint(&mut self, identifier: &str) -> Result<Self::Committed> {
        Ok(Self::Committed::new(
            identifier,
            &self.committed_path.join(identifier),
        ))
    }

    fn remove_checkpoint(&mut self, identifier: &str) -> Result<()> {
        let committed_checkpoint_path = self.committed_path.join(identifier);
        fs::remove_dir_all(committed_checkpoint_path)
            .map_err(|_| Error::io(format!("Could not remove checkpoint: {}", identifier)))?;

        Ok(())
    }

    fn checkpoint_identifiers(&mut self) -> Result<Vec<String>> {
        let mut identifiers = Vec::new();
        let entries = fs::read_dir(&self.committed_path)
            .map_err(|_| Error::io("Could not retrieve checkpoint identifiers".to_string()))?;

        for entry in entries {
            let entry = entry
                .map_err(|_| Error::io("Could not retrieve checkpoint identifiers".to_string()))?;

            let path = entry.path();
            if path.is_dir() {
                if let Some(file_name) = path.file_name() {
                    if let Some(file_name) = file_name.to_str() {
                        identifiers.push(file_name.to_string());
                    } else {
                        Err(Error::invalid_identifier(format!(
                            "Directory {:?} contains invalid checkpoint identifier",
                            self.committed_path
                        )))?;
                    }
                }
            }
        }

        Ok(identifiers)
    }
}

// Provides access to checkpoint data stored in the file system.
#[derive(Debug)]
struct CheckpointStorage {
    identifier: String,
    path: PathBuf,
}

impl CheckpointStorage {
    fn new(identifier: &str, path: &Path) -> Self {
        Self {
            identifier: identifier.to_string(),
            path: path.to_path_buf(),
        }
    }
}

impl CheckpointStorage {
    fn get<T: DeserializeOwned + Serialize>(&mut self, key: &str) -> Result<T> {
        let filepath = self.path.join(key);
        let identifier = self.identifier()?;
        if !filepath.exists() {
            Err(Error::cp_key_doesnt_exist(format!(
                "No value stored for checkpoint: {}, key: {}",
                identifier, key
            )))?;
        }

        let file = File::open(filepath).map_err(|_| {
            Error::io(format!(
                "Could not retrieve value for checkpoint: {}, key: {}",
                identifier, key
            ))
        })?;

        let reader = BufReader::new(file);
        let value = bincode::deserialize_from(reader).map_err(|_| {
            Error::deserialize(format!(
                "Failed to deserialize value from checkpoint: {}, key: {}",
                identifier, key
            ))
        })?;

        Ok(value)
    }

    fn keys(&mut self) -> Result<Vec<String>> {
        let identifier = self.identifier()?.to_string();
        let entries = fs::read_dir(&self.path).map_err(|_| {
            Error::io(format!(
                "Could not retrieve keys for checkpoint: {}",
                &identifier
            ))
        })?;

        let mut keys = Vec::new();
        for entry in entries {
            let entry = entry.map_err(|_| {
                Error::io(format!(
                    "Could not retrieve keys for checkpoint: {}",
                    &identifier
                ))
            })?;

            let path = entry.path();
            if path.is_file() {
                if let Some(file_name) = path.file_name() {
                    if let Some(file_name) = file_name.to_str() {
                        keys.push(file_name.to_string());
                    } else {
                        Err(Error::invalid_identifier(format!(
                            "Directory {:?} contains invalid checkpoint identifier",
                            self.path
                        )))?;
                    }
                }
            }
        }

        Ok(keys)
    }

    fn identifier(&mut self) -> Result<&str> {
        Ok(self.identifier.as_str())
    }

    fn put<T: DeserializeOwned + Serialize>(&mut self, key: &str, value: &T) -> Result<()> {
        let identifier = self.identifier()?.to_string();
        let filepath = self.path.join(key);
        let file = File::create(filepath).map_err(|_| {
            Error::io(format!(
                "Could not insert value into checkpoint: {}, key: {}",
                &identifier, key
            ))
        })?;

        let writer = BufWriter::new(file);
        bincode::serialize_into(writer, value).map_err(|_| {
            Error::serialize(format!(
                "Failed to serialize value into checkpoint: {}, key: {}",
                identifier, key
            ))
        })?;
        Ok(())
    }

    fn remove(&mut self, key: &str) -> Result<()> {
        let identifier = self.identifier()?.to_string();
        let filepath = self.path.join(key);
        if !filepath.exists() {
            Err(Error::cp_key_doesnt_exist(format!(
                "No value to remove for checkpoint: {}, key: {}",
                &identifier, key
            )))?;
        }

        fs::remove_file(filepath).map_err(|_| {
            Error::io(format!(
                "Could not remove value from checkpoint: {}, key: {}",
                &identifier, key
            ))
        })?;
        Ok(())
    }
}

/// Provides access to committed checkpoint data stored in the file system.
#[derive(Debug)]
pub struct Committed {
    inner: CheckpointStorage,
}

impl Committed {
    fn new(identifier: &str, path: &Path) -> Self {
        Self {
            inner: CheckpointStorage::new(identifier, path),
        }
    }
}

impl CommittedCheckpoint for Committed {
    fn get<T: DeserializeOwned + Serialize>(&mut self, key: &str) -> Result<T> {
        self.inner.get(key)
    }

    fn keys(&mut self) -> Result<Vec<String>> {
        self.inner.keys()
    }

    fn identifier(&mut self) -> Result<&str> {
        self.inner.identifier()
    }
}

/// Provides access to uncommitted checkpoint data stored in the file system.
#[derive(Debug)]
pub struct Uncommitted {
    inner: CheckpointStorage,
}

impl Uncommitted {
    fn new(identifier: &str, path: &Path) -> Self {
        Self {
            inner: CheckpointStorage::new(identifier, path),
        }
    }
}

impl UncommittedCheckpoint for Uncommitted {
    fn get<T: DeserializeOwned + Serialize>(&mut self, key: &str) -> Result<T> {
        self.inner.get(key)
    }

    fn keys(&mut self) -> Result<Vec<String>> {
        self.inner.keys()
    }

    fn identifier(&mut self) -> Result<&str> {
        self.inner.identifier()
    }

    fn put<T: DeserializeOwned + Serialize>(&mut self, key: &str, value: &T) -> Result<()> {
        self.inner.put(key, value)
    }

    fn remove(&mut self, key: &str) -> Result<()> {
        self.inner.remove(key)
    }
}

impl Drop for Uncommitted {
    fn drop(&mut self) {
        if self.inner.path.exists() {
            let _ = fs::remove_dir_all(&self.inner.path);
        }
    }
}

#[cfg(all(test, feature = "filestorage-tests"))]
mod test {
    use std::fs;
    use std::mem;

    use error::ErrorKind;
    use storage::{CommittedCheckpoint, FileStorage, Storage, UncommittedCheckpoint};
    use test_utils::{create_named_test_file, create_test_directory,
                     create_test_directory_with_checkpoint_folders};

    // Verify that opening a FileStorage object with an existing path works as expected.
    #[test]
    fn file_storage_open_existing_path() {
        let dir = create_test_directory();
        let path = dir.path();
        let committed_path = path.join("committed");
        let uncommitted_path = path.join("uncommitted");

        // Verify that FileStorage::open() returns an error when a filepath is provided.
        let temp_file = create_named_test_file(&path);
        assert_eq!(
            FileStorage::open(&temp_file.path()).unwrap_err().kind(),
            ErrorKind::Initialization
        );

        // Verify that FileStorage::open() returns an error when the specified directory does not
        // contain a "committed" sub-directory.
        fs::create_dir(&uncommitted_path).unwrap();
        assert_eq!(
            FileStorage::open(path).unwrap_err().kind(),
            ErrorKind::Initialization
        );
        fs::remove_dir(&uncommitted_path).unwrap();

        // Verify that FileStorage::open() returns an error when the specified directory does not
        // contain an "uncommitted" sub-directory.
        fs::create_dir(&committed_path).unwrap();
        assert_eq!(
            FileStorage::open(path).unwrap_err().kind(),
            ErrorKind::Initialization
        );
        fs::remove_dir(&committed_path).unwrap();

        // Verify that FileStorage::open() is successful when the specified directory contains both
        // a "committed" sub-directory and an "uncommitted" sub-directory.
        fs::create_dir(&committed_path).unwrap();
        fs::create_dir(&uncommitted_path).unwrap();
        assert!(FileStorage::open(path).is_ok());
    }

    // Verify that opening a FileStorage object with a non-existing path works as expected.
    #[test]
    fn file_storage_open_non_existing_path() {
        let dir = create_test_directory();

        // Add "Checkpoints" to the path so that an empty directory can be provided to
        // FileStorage::open() below.
        let path = dir.path().join("Checkpoints");

        // Verify that FileStorage::open() is successful when the specified directory does
        // not exist.
        assert!(FileStorage::open(&path).is_ok());
    }

    // Verify that checkpoint creation and commit behave as expected.
    #[test]
    fn file_storage_create_and_commit_checkpoint() {
        let dir = create_test_directory_with_checkpoint_folders();
        let path = dir.path();

        let mut storage = FileStorage::open(&path).unwrap();

        // Verify that create_checkpoint() can be called twice with the same checkpoint identifier
        // as long as the uncommitted checkpoint returned from the first call is dropped before
        // the second call.
        assert!(storage.create_checkpoint("CP").is_ok());
        assert!(storage.create_checkpoint("CP").is_ok());

        // Verify that create_checkpoint() returns an error when a checkpoint creation is attempted
        // with the same identifier as an uncommitted checkpoint.
        let uncommitted_cp = storage.create_checkpoint("CP").unwrap();
        assert_eq!(
            storage.create_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpAlreadyExists
        );

        // Verify that create_checkpoint() returns an error when a checkpoint creation is attempted
        // with the same identifier as a committed checkpoint that is in use (i.e. has a variable
        // binding).
        let committed_cp = storage.commit_checkpoint(uncommitted_cp).unwrap();
        assert_eq!(
            storage.create_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpAlreadyExists
        );

        // Verify that create_checkpoint() returns an error when a checkpoint creation is attempted
        // with the same identifier as a committed checkpoint that is not in use (i.e. has no
        // variable binding).
        mem::drop(committed_cp);
        assert_eq!(
            storage.create_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpAlreadyExists
        );
    }

    // Verify that checkpoint loading and removal behave as expected.
    #[test]
    fn file_storage_load_and_remove_checkpoint() {
        let dir = create_test_directory_with_checkpoint_folders();
        let path = dir.path();

        let mut storage = FileStorage::open(&path).unwrap();

        // Verify that load_checkpoint() returns an error when a checkpoint load is attempted but
        // a committed checkpoint with the given identifier does not exist.
        assert_eq!(
            storage.load_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpDoesntExist
        );

        // Verify that load_checkpoint() returns an error when a checkpoint load is attempted, even
        // if an uncommitted checkpoint with the given identifier exists.
        let uncommitted_cp = storage.create_checkpoint("CP").unwrap();
        assert_eq!(
            storage.load_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpDoesntExist
        );

        // Verify that load_checkpoint() returns a committed checkpoint binding when a committed
        // checkpoint with the given identifier exists and is in use (i.e. has a variable binding).
        let committed_cp = storage.commit_checkpoint(uncommitted_cp).unwrap();
        assert!(storage.load_checkpoint("CP").is_ok());

        // Verify that load_checkpoint() returns a committed checkpoint binding when a committed
        // checkpoint with the given identifier exists and is not in use (i.e. has no
        // variable binding).
        mem::drop(committed_cp);
        let result = storage.load_checkpoint("CP");
        assert!(result.is_ok());

        // Verify that remove_checkpoint() returns an error when an attempt to remove a checkpoint
        // that is in use (i.e. has a variable binding) is made.
        let committed_cp = result.unwrap();
        assert_eq!(
            storage.remove_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpInUse
        );

        // Verify that remove_checkpoint() is successful when an attempt to remove a checkpoint
        // that is not in use (i.e. does not have a variable binding) is made.
        mem::drop(committed_cp);
        assert!(storage.remove_checkpoint("CP").is_ok());

        // Verify that remove_checkpoint() returns an error when an attempt to remove a checkpoint
        // that doesn't exist is made.
        assert_eq!(
            storage.remove_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpDoesntExist
        );

        // Verify that the mechanism which keeps track of the number of times a committed checkpoint
        // has been loaded works as expected.
        let uncommitted_cp = storage.create_checkpoint("CP").unwrap();
        let cp1 = storage.commit_checkpoint(uncommitted_cp).unwrap();
        assert_eq!(
            storage.remove_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpInUse
        );

        let cp2 = storage.load_checkpoint("CP").unwrap();
        assert_eq!(
            storage.remove_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpInUse
        );

        mem::drop(cp1);
        assert_eq!(
            storage.remove_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpInUse
        );

        let cp3 = storage.load_checkpoint("CP").unwrap();
        assert_eq!(
            storage.remove_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpInUse
        );

        let cp4 = storage.load_checkpoint("CP").unwrap();
        assert_eq!(
            storage.remove_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpInUse
        );

        let cp5 = storage.load_checkpoint("CP").unwrap();
        assert_eq!(
            storage.remove_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpInUse
        );

        // Start dropping loaded checkpoints in an arbitrary order.
        mem::drop(cp4);
        assert_eq!(
            storage.remove_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpInUse
        );

        mem::drop(cp2);
        assert_eq!(
            storage.remove_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpInUse
        );

        mem::drop(cp5);
        assert_eq!(
            storage.remove_checkpoint("CP").unwrap_err().kind(),
            ErrorKind::CpInUse
        );

        mem::drop(cp3);
        assert!(storage.remove_checkpoint("CP").is_ok());
    }

    // Verify that checkpoint identifiers are returned properly.
    #[test]
    fn file_storage_checkpoint_identifiers() {
        let dir = create_test_directory_with_checkpoint_folders();
        let path = dir.path();

        let mut storage = FileStorage::open(&path).unwrap();

        // Verify that checkpoint_identifiers() returns an empty list when there are no
        // committed checkpoints.
        assert_eq!(
            storage.checkpoint_identifiers().unwrap(),
            Vec::<String>::new()
        );

        // Verify that checkpoint_identifiers() returns an empty list even when there are
        // uncommitted checkpoints.
        let uncommitted_cp_1 = storage.create_checkpoint("CP1").unwrap();
        assert_eq!(
            storage.checkpoint_identifiers().unwrap(),
            Vec::<String>::new()
        );

        // Verify that checkpoint_identifiers() returns a list of identifiers when committed
        // checkpoints exist and are loaded (i.e. when they have variable bindings).
        let committed_cp_1 = storage.commit_checkpoint(uncommitted_cp_1).unwrap();
        assert_eq!(
            storage.checkpoint_identifiers().unwrap(),
            vec!["CP1".to_string()]
        );

        let uncommitted_cp_2 = storage.create_checkpoint("CP2").unwrap();
        let committed_cp_2 = storage.commit_checkpoint(uncommitted_cp_2).unwrap();
        let mut identifiers = storage.checkpoint_identifiers().unwrap();
        identifiers.sort();
        assert_eq!(identifiers, vec!["CP1".to_string(), "CP2".to_string()]);

        let uncommitted_cp_3 = storage.create_checkpoint("CP3").unwrap();
        let committed_cp_3 = storage.commit_checkpoint(uncommitted_cp_3).unwrap();
        let mut identifiers = storage.checkpoint_identifiers().unwrap();
        identifiers.sort();
        assert_eq!(
            identifiers,
            vec!["CP1".to_string(), "CP2".to_string(), "CP3".to_string()]
        );

        // Verify that checkpoint_identifiers() returns a list of identifiers when committed
        // checkpoints exist and are not loaded (i.e. when they do not have variable bindings).
        mem::drop(committed_cp_1);
        mem::drop(committed_cp_2);
        mem::drop(committed_cp_3);

        let mut identifiers = storage.checkpoint_identifiers().unwrap();
        identifiers.sort();
        assert_eq!(
            identifiers,
            vec!["CP1".to_string(), "CP2".to_string(), "CP3".to_string()]
        );

        // Verify that when storage is opened with an existing path that checkpoint identifiers are
        // returned properly.
        drop(storage);

        let mut storage = FileStorage::open(&path).unwrap();
        let mut identifiers = storage.checkpoint_identifiers().unwrap();
        identifiers.sort();
        assert_eq!(
            identifiers,
            vec!["CP1".to_string(), "CP2".to_string(), "CP3".to_string()]
        );

        // Verify that checkpoint identifiers are removed from the list of identifiers returned by
        // checkpoint_identifiers() when the associated checkpoint is removed.
        storage.remove_checkpoint("CP2").unwrap();
        let mut identifiers = storage.checkpoint_identifiers().unwrap();
        identifiers.sort();
        assert_eq!(identifiers, vec!["CP1".to_string(), "CP3".to_string()]);

        storage.remove_checkpoint("CP1").unwrap();
        let mut identifiers = storage.checkpoint_identifiers().unwrap();
        identifiers.sort();
        assert_eq!(identifiers, vec!["CP3".to_string()]);

        storage.remove_checkpoint("CP3").unwrap();
        let mut identifiers = storage.checkpoint_identifiers().unwrap();
        identifiers.sort();
        assert_eq!(identifiers, Vec::<String>::new());
    }

    // Verify that uncommitted and committed checkpoints behave as expected.
    #[test]
    fn uncommitted_and_committed() {
        let dir = create_test_directory_with_checkpoint_folders();
        let path = dir.path();

        let mut storage = FileStorage::open(&path).unwrap();

        // Verify that the methods of an empty uncommitted checkpoint behave as expected.
        let mut uncommitted_cp_1 = storage.create_checkpoint("CP1").unwrap();
        assert_eq!(uncommitted_cp_1.identifier().unwrap(), "CP1");
        assert_eq!(uncommitted_cp_1.keys().unwrap(), Vec::<String>::new());
        assert_eq!(
            uncommitted_cp_1.get::<u8>("Key1").unwrap_err().kind(),
            ErrorKind::CpKeyDoesntExist
        );
        assert_eq!(
            uncommitted_cp_1.remove("Key1").unwrap_err().kind(),
            ErrorKind::CpKeyDoesntExist
        );

        // Add data to the uncommitted checkpoint.
        let data_1 = 0u8;
        let data_2 = String::from("Test String.");
        let data_3 = Some(true);
        assert!(uncommitted_cp_1.put("Key1", &data_1).is_ok());
        assert!(uncommitted_cp_1.put("Key2", &data_2).is_ok());
        assert!(uncommitted_cp_1.put("Key3", &data_3).is_ok());

        // Verify that adding and removing values to existing checkpoints works as expected and
        // that creating a new uncommitted checkpoint does not affect the data stored in another
        // uncommitted checkpoint.
        let mut uncommitted_cp_2 = storage.create_checkpoint("CP2").unwrap();
        assert_eq!(uncommitted_cp_2.identifier().unwrap(), "CP2");
        assert_eq!(uncommitted_cp_2.keys().unwrap(), Vec::<String>::new());
        assert_eq!(
            uncommitted_cp_2.get::<u8>("Key1").unwrap_err().kind(),
            ErrorKind::CpKeyDoesntExist
        );
        assert_eq!(
            uncommitted_cp_2.remove("Key1").unwrap_err().kind(),
            ErrorKind::CpKeyDoesntExist
        );

        assert_eq!(uncommitted_cp_1.identifier().unwrap(), "CP1");
        let mut keys = uncommitted_cp_1.keys().unwrap();
        keys.sort();
        assert_eq!(keys, vec!["Key1", "Key2", "Key3"]);
        assert_eq!(uncommitted_cp_1.get::<u8>("Key1").unwrap(), data_1);
        assert_eq!(uncommitted_cp_1.get::<String>("Key2").unwrap(), data_2);
        assert_eq!(
            uncommitted_cp_1.get::<Option<bool>>("Key3").unwrap(),
            data_3
        );
        assert!(uncommitted_cp_1.remove("Key2").is_ok());
        assert_eq!(
            uncommitted_cp_1.remove("Key2").unwrap_err().kind(),
            ErrorKind::CpKeyDoesntExist
        );
        let mut keys = uncommitted_cp_1.keys().unwrap();
        keys.sort();
        assert_eq!(keys, vec!["Key1", "Key3"]);
        assert!(uncommitted_cp_1.remove("Key1").is_ok());
        assert_eq!(
            uncommitted_cp_1.remove("Key1").unwrap_err().kind(),
            ErrorKind::CpKeyDoesntExist
        );
        let mut keys = uncommitted_cp_1.keys().unwrap();
        keys.sort();
        assert_eq!(keys, vec!["Key3"]);
        assert!(uncommitted_cp_1.remove("Key3").is_ok());
        assert_eq!(
            uncommitted_cp_1.remove("Key3").unwrap_err().kind(),
            ErrorKind::CpKeyDoesntExist
        );
        assert_eq!(uncommitted_cp_1.keys().unwrap(), Vec::<String>::new());

        // Verify that the value for a given key can be added even if the key was previously
        // removed.
        assert!(uncommitted_cp_1.put("Key1", &data_1).is_ok());
        assert!(uncommitted_cp_1.put("Key2", &data_2).is_ok());
        assert!(uncommitted_cp_1.put("Key3", &data_3).is_ok());
        assert_eq!(uncommitted_cp_1.identifier().unwrap(), "CP1");
        let mut keys = uncommitted_cp_1.keys().unwrap();
        keys.sort();
        assert_eq!(keys, vec!["Key1", "Key2", "Key3"]);
        assert_eq!(uncommitted_cp_1.get::<u8>("Key1").unwrap(), data_1);
        assert_eq!(uncommitted_cp_1.get::<String>("Key2").unwrap(), data_2);
        assert_eq!(
            uncommitted_cp_1.get::<Option<bool>>("Key3").unwrap(),
            data_3
        );

        // Verify that the value for a given key can be changed.
        let data_1 = String::from("Test String.");
        let data_2 = Some(true);
        let data_3 = 0u8;
        assert!(uncommitted_cp_1.put("Key1", &data_1).is_ok());
        assert!(uncommitted_cp_1.put("Key2", &data_2).is_ok());
        assert!(uncommitted_cp_1.put("Key3", &data_3).is_ok());

        let mut keys = uncommitted_cp_1.keys().unwrap();
        keys.sort();
        assert_eq!(keys, vec!["Key1", "Key2", "Key3"]);
        assert_eq!(uncommitted_cp_1.get::<String>("Key1").unwrap(), data_1);
        assert_eq!(
            uncommitted_cp_1.get::<Option<bool>>("Key2").unwrap(),
            data_2
        );
        assert_eq!(uncommitted_cp_1.get::<u8>("Key3").unwrap(), data_3);

        assert!(uncommitted_cp_1.put("Key1", &data_2).is_ok());
        assert!(uncommitted_cp_1.put("Key2", &data_3).is_ok());
        assert!(uncommitted_cp_1.put("Key3", &data_1).is_ok());
        assert_eq!(uncommitted_cp_1.identifier().unwrap(), "CP1");
        let mut keys = uncommitted_cp_1.keys().unwrap();
        keys.sort();
        assert_eq!(keys, vec!["Key1", "Key2", "Key3"]);
        assert_eq!(
            uncommitted_cp_1.get::<Option<bool>>("Key1").unwrap(),
            data_2
        );
        assert_eq!(uncommitted_cp_1.get::<u8>("Key2").unwrap(), data_3);
        assert_eq!(uncommitted_cp_1.get::<String>("Key3").unwrap(), data_1);

        // Verify that uncommitted checkpoints contain the same values after being committed.
        let mut committed_cp_1 = storage.commit_checkpoint(uncommitted_cp_1).unwrap();
        let mut committed_cp_2 = storage.commit_checkpoint(uncommitted_cp_2).unwrap();

        assert_eq!(committed_cp_1.identifier().unwrap(), "CP1");
        let mut keys = committed_cp_1.keys().unwrap();
        keys.sort();
        assert_eq!(keys, vec!["Key1", "Key2", "Key3"]);
        assert_eq!(committed_cp_1.get::<Option<bool>>("Key1").unwrap(), data_2);
        assert_eq!(committed_cp_1.get::<u8>("Key2").unwrap(), data_3);
        assert_eq!(committed_cp_1.get::<String>("Key3").unwrap(), data_1);
        assert_eq!(
            committed_cp_1.get::<u8>("Key4").unwrap_err().kind(),
            ErrorKind::CpKeyDoesntExist
        );

        assert_eq!(committed_cp_2.identifier().unwrap(), "CP2");
        assert_eq!(committed_cp_2.keys().unwrap(), Vec::<String>::new());
        assert_eq!(
            committed_cp_2.get::<u8>("Key1").unwrap_err().kind(),
            ErrorKind::CpKeyDoesntExist
        );
    }
}