Blocktree/crates/btfuse/src/main.rs

// SPDX-License-Identifier: AGPL-3.0-or-later
mod fuse_daemon;
use btconfig::{config_keys, ConfigBuilderExt, CredStoreCfg, NodeCredConsumer};
use fuse_daemon::FuseDaemon;
mod fuse_fs;

use ::config::Config as ExtConfig;
use btfproto::{client::FsClient, local_fs::LocalFs, server::FsProvider};
use btlib::{
    crypto::{Creds, CredsPriv},
    error::BtErr,
    Result,
};
use btmsg::{transmitter, BlockAddr};
use serde::Deserialize;
use std::{
    fs::{self},
    io,
    num::NonZeroUsize,
    path::{Path, PathBuf},
    sync::Arc,
};
use tokio::sync::oneshot;

#[derive(Debug, PartialEq, Eq, Clone, Deserialize)]
#[serde(tag = "type")]
pub enum FsKind {
    Local { path: PathBuf },
    Remote { addr: BlockAddr },
}

impl FsKind {
    config_keys! {
        const STRUCT_KEY = "fskind";
        const TAG = "type";
        const PATH = "path";
        const ADDR = "addr";
    }
}

#[derive(Debug, Clone, Deserialize)]
struct AppConfig {
    credstore: CredStoreCfg,
    fskind: FsKind,
    mntdir: PathBuf,
    mntoptions: String,
    threads: Option<NonZeroUsize>,
}

impl AppConfig {
    fn new() -> Result<Self> {
        ExtConfig::builder()
            .set_default(FsKind::TAG, "Local")?
            .set_default(FsKind::PATH, "./state/blocks")?
            .set_default("mntdir", "./state/mnt")?
            .set_default("mntoptions", "default_permissions")?
            .btconfig()?
            .build()?
            .try_deserialize()
            .bterr()
    }
}

trait PathExt {
    fn try_create_dir(&self) -> io::Result<()>;
}

impl<T: AsRef<Path>> PathExt for T {
    fn try_create_dir(&self) -> io::Result<()> {
        match fs::create_dir(self) {
            Ok(_) => Ok(()),
            Err(err) => match err.kind() {
                io::ErrorKind::AlreadyExists => Ok(()),
                _ => Err(err),
            },
        }
    }
}

async fn local_provider(btdir: PathBuf, node_creds: Arc<dyn Creds>) -> Result<impl FsProvider> {
    btdir.try_create_dir()?;
    let empty = fs::read_dir(&btdir)?.next().is_none();
    if empty {
        LocalFs::new_empty(btdir, 0, node_creds, btfproto::local_fs::ModeAuthorizer {}).await
    } else {
        LocalFs::new_existing(btdir, node_creds, btfproto::local_fs::ModeAuthorizer {})
    }
}

async fn remote_provider<C: 'static + Creds + Send + Sync>(
    remote_addr: BlockAddr,
    node_creds: C,
) -> Result<impl FsProvider> {
    let tx = transmitter(Arc::new(remote_addr), Arc::new(node_creds)).await?;
    let client = FsClient::new(tx);
    Ok(client)
}

async fn run_daemon(config: AppConfig, mounted_signal: Option<oneshot::Sender<()>>) {
    let node_creds = config.credstore.consume(NodeCredConsumer).unwrap().unwrap();
    let fallback_path = {
        let writecap = node_creds
            .writecap()
            .ok_or(btlib::BlockError::MissingWritecap)
            .unwrap();
        Arc::new(writecap.bind_path())
    };
    let mut daemon = match config.fskind {
        FsKind::Local { path: btdir } => {
            log::info!("starting daemon with local provider using {:?}", btdir);
            let provider = local_provider(btdir, node_creds)
                .await
                .expect("failed to create local provider");
            FuseDaemon::new(
                config.mntdir,
                &config.mntoptions,
                config.threads,
                fallback_path,
                mounted_signal,
                provider,
            )
        }
        FsKind::Remote { addr: remote_addr } => {
            log::info!(
                "starting daemon with remote provider using {:?}",
                remote_addr.socket_addr()
            );
            let provider = remote_provider(remote_addr, node_creds)
                .await
                .expect("failed to create remote provider");
            FuseDaemon::new(
                config.mntdir,
                &config.mntoptions,
                config.threads,
                fallback_path,
                mounted_signal,
                provider,
            )
        }
    }
    .expect("failed to create FUSE daemon");

    daemon.finished().await;
}

#[tokio::main]
async fn main() {
    env_logger::init();
    let config = AppConfig::new().unwrap();
    run_daemon(config, None).await;
}

#[cfg(test)]
mod test {
    use super::*;

    use btfproto::{local_fs::ModeAuthorizer, server::new_fs_server};
    use btlib::{
        crypto::{tpm::TpmCredStore, CredStore, CredStoreMut, Creds},
        log::BuilderExt,
        Epoch, Principaled,
    };
    use btmsg::Receiver;
    use ctor::ctor;
    use std::{
        ffi::{OsStr, OsString},
        fs::Permissions,
        io::{BufRead, SeekFrom},
        net::{IpAddr, Ipv6Addr},
        num::NonZeroUsize,
        os::unix::fs::PermissionsExt,
        time::Duration,
    };
    use swtpm_harness::SwtpmHarness;
    use tempdir::TempDir;
    use tokio::{
        fs::{
            create_dir, hard_link, metadata, read, read_dir, remove_dir, remove_file, rename,
            set_permissions, write, OpenOptions, ReadDir,
        },
        io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt},
        task::JoinHandle,
    };

    /// An optional timeout to limit the time spent waiting for the FUSE daemon to start in tests.
    const TIMEOUT: Option<Duration> = Some(Duration::from_millis(1250));

    /// The log level to use when running tests.
    /// Note that the debug log level significantly reduces performance.
    const LOG_LEVEL: &str = "warn";

    #[ctor]
    fn ctor() {
        std::env::set_var("RUST_LOG", LOG_LEVEL);
        env_logger::Builder::from_default_env().btformat().init();
    }

    /// Reads `/etc/mtab` to determine if `mnt_path` is mounted. Returns true if it is and false
    /// otherwise.
    fn mounted(mnt_path: &str) -> bool {
        let file = std::fs::OpenOptions::new()
            .read(true)
            .write(false)
            .create(false)
            .open("/etc/mtab")
            .unwrap();
        let mut reader = std::io::BufReader::new(file);
        let mut line = String::with_capacity(64);
        loop {
            line.clear();
            let read = reader.read_line(&mut line).unwrap();
            if 0 == read {
                break;
            }
            let path = line.split(' ').skip(1).next().unwrap();
            if path == mnt_path {
                return true;
            }
        }
        false
    }

    /// Unmounts the file system at the given path.
    fn unmount<P: AsRef<Path>>(mnt_path: P) {
        let mnt_path = mnt_path.as_ref();
        if !mounted(mnt_path.to_str().unwrap()) {
            return;
        }
        const PROG: &str = "fusermount";
        let mnt_path = mnt_path
            .as_os_str()
            .to_str()
            .expect("failed to convert mnt_path to `str`");
        let code = std::process::Command::new(PROG)
            .args(["-z", "-u", mnt_path])
            .status()
            .expect("waiting for exit status failed")
            .code()
            .expect("code returned None");
        if code != 0 {
            panic!("{PROG} exited with a non-zero status: {code}");
        }
    }

    async fn file_names(mut read_dir: ReadDir) -> Vec<OsString> {
        let mut output = Vec::new();
        while let Some(entry) = read_dir.next_entry().await.unwrap() {
            output.push(entry.file_name());
        }
        output
    }

    const ROOT_PASSWD: &str = "password";

    struct TestCase<R: Receiver> {
        config: AppConfig,
        handle: Option<JoinHandle<()>>,
        node_principal: OsString,
        stop_flag: Option<()>,
        // Note that the drop order of these fields is significant.
        _receiver: Option<R>,
        _cred_store: TpmCredStore,
        _swtpm: SwtpmHarness,
        _temp_dir: TempDir,
    }

    async fn new_local() -> TestCase<impl Receiver> {
        new(false).await
    }

    async fn new_remote() -> TestCase<impl Receiver> {
        new(true).await
    }

    async fn new(remote: bool) -> TestCase<impl Receiver> {
        let tmp = TempDir::new("btfuse").unwrap();
        let (mounted_tx, mounted_rx) = oneshot::channel();
        let (swtpm, cred_store) = swtpm();
        let block_dir = tmp.path().join("bt");
        let (fs_kind, receiver) = if remote {
            let node_creds = Arc::new(cred_store.node_creds().unwrap());
            let bind_path = node_creds.bind_path().unwrap();
            block_dir.try_create_dir().unwrap();
            let local_fs = LocalFs::new_empty(block_dir, 0, node_creds.clone(), ModeAuthorizer)
                .await
                .unwrap();
            let ip_addr = IpAddr::V6(Ipv6Addr::LOCALHOST);
            let receiver = new_fs_server(ip_addr, node_creds.clone(), Arc::new(local_fs)).unwrap();
            let fs_kind = FsKind::Remote {
                addr: BlockAddr::new(ip_addr, Arc::new(bind_path)),
            };
            (fs_kind, Some(receiver))
        } else {
            (FsKind::Local { path: block_dir }, None)
        };
        let config = AppConfig {
            threads: Some(NonZeroUsize::new(1).unwrap()),
            mntdir: tmp.path().join("mnt"),
            credstore: CredStoreCfg::Tpm {
                path: swtpm.state_path().to_owned().into(),
                tabrmd: swtpm.tabrmd_config().to_owned(),
            },
            fskind: fs_kind,
            mntoptions: "default_permissions".to_string(),
        };
        let config_clone = config.clone();
        let handle = tokio::spawn(async move {
            run_daemon(config_clone, Some(mounted_tx)).await;
        });
        if let Some(timeout) = TIMEOUT {
            tokio::time::timeout(timeout, mounted_rx)
                .await
                .unwrap()
                .unwrap();
        } else {
            mounted_rx.await.unwrap();
        }
        let node_principal =
            OsString::from(cred_store.node_creds().unwrap().principal().to_string());
        TestCase {
            config,
            handle: Some(handle),
            node_principal,
            stop_flag: Some(()),
            _receiver: receiver,
            _temp_dir: tmp,
            _swtpm: swtpm,
            _cred_store: cred_store,
        }
    }

    fn swtpm() -> (SwtpmHarness, TpmCredStore) {
        let swtpm = SwtpmHarness::new().unwrap();
        let state_path: PathBuf = swtpm.state_path().to_owned();
        let cred_store = {
            let context = swtpm.context().unwrap();
            TpmCredStore::from_context(context, state_path.clone()).unwrap()
        };
        let root_creds = cred_store.gen_root_creds(ROOT_PASSWD).unwrap();
        let mut node_creds = cred_store.node_creds().unwrap();
        let expires = Epoch::now() + Duration::from_secs(3600);
        let writecap = root_creds
            .issue_writecap(node_creds.principal(), vec![], expires)
            .unwrap();
        cred_store
            .assign_node_writecap(&mut node_creds, writecap)
            .unwrap();
        (swtpm, cred_store)
    }

    impl<R: Receiver> TestCase<R> {
        fn mnt_dir(&self) -> &Path {
            &self.config.mntdir
        }

        /// Signals to the daemon that it must stop.
        fn signal_stop(&mut self) {
            if let Some(_) = self.stop_flag.take() {
                unmount(&self.config.mntdir)
            }
        }

        /// Returns a future that resolves when the daemon has stopped.
        async fn stopped(&mut self) {
            if let Some(handle) = self.handle.take() {
                handle.await.expect("join failed");
            }
        }

        /// Signals to the daemon to stop and returns a future that resolves when after it has
        /// stopped.
        async fn stop(&mut self) {
            self.signal_stop();
            self.stopped().await;
        }

        fn initial_contents(&self) -> Vec<&OsStr> {
            vec![&self.node_principal]
        }
    }

    impl<R: Receiver> Drop for TestCase<R> {
        fn drop(&mut self) {
            self.signal_stop()
        }
    }

    /// Creates a new file system and mounts it at `/tmp/btfuse.<random>/mnt` so it can be
    /// tested manually.
    //#[tokio::test]
    #[allow(dead_code)]
    async fn manual_test() {
        let mut case = new_local().await;
        case.stopped().await
    }

    async fn write_read(mut case: TestCase<impl Receiver>) -> Result<()> {
        const EXPECTED: &[u8] =
            b"The paths to failure are uncountable, yet to success there are few.";
        let file_path = case.mnt_dir().join("file");

        write(&file_path, EXPECTED).await?;

        let actual = read(&file_path).await?;
        assert_eq!(EXPECTED, actual);

        case.stop().await;
        Ok(())
    }

    #[tokio::test]
    async fn write_read_local() -> Result<()> {
        write_read(new_local().await).await
    }

    // When the current thread runtime is used the test executable does not exit after the test
    // method returns because one of the FuseDaemon blocking threads is blocked calling
    // `FuseChannel::get_request`.
    #[tokio::test(flavor = "multi_thread")]
    async fn write_read_remote() -> Result<()> {
        write_read(new_remote().await).await
    }

    async fn create_file_then_readdir(mut case: TestCase<impl Receiver>) {
        const DATA: &[u8] = b"Au revoir Shoshanna!";
        let file_name = OsStr::new("landa_dialog.txt");
        let mut expected = case.initial_contents();
        expected.push(file_name);
        let mnt_path = case.mnt_dir();
        let file_path = mnt_path.join(file_name);

        write(&file_path, DATA).await.expect("write failed");

        let first = file_names(read_dir(&mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(expected, first);
        let second = file_names(read_dir(&mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(expected, second);

        case.stop().await;
    }

    #[tokio::test]
    async fn create_file_then_readdir_local() {
        create_file_then_readdir(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn create_file_then_readdir_remote() {
        create_file_then_readdir(new_remote().await).await
    }

    async fn create_then_delete_file(mut case: TestCase<impl Receiver>) {
        const DATA: &[u8] = b"The universe is hostile, so impersonal. Devour to survive";
        let file_name = OsStr::new("tool_lyrics.txt");
        let mnt_path = case.mnt_dir();
        let file_path = mnt_path.join(file_name);
        write(&file_path, DATA).await.expect("write failed");

        remove_file(&file_path).await.expect("remove_file failed");

        let expected = case.initial_contents();
        let actual = file_names(read_dir(&mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(expected, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn create_then_delete_file_local() {
        create_then_delete_file(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn create_then_delete_file_remote() {
        create_then_delete_file(new_remote().await).await
    }

    async fn hard_link_then_remove(mut case: TestCase<impl Receiver>) {
        const EXPECTED: &[u8] = b"And the lives we've reclaimed";
        let name1 = OsStr::new("refugee_lyrics.txt");
        let name2 = OsStr::new("rise_against_lyrics.txt");
        let mnt_path = case.mnt_dir();
        let path1 = mnt_path.join(name1);
        let path2 = mnt_path.join(name2);
        write(&path1, EXPECTED).await.expect("write failed");

        hard_link(&path1, &path2).await.expect("hard_link failed");
        remove_file(&path1).await.expect("remove_file failed");

        let actual = read(&path2).await.expect("read failed");
        assert_eq!(EXPECTED, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn hard_link_then_remove_local() {
        hard_link_then_remove(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn hard_link_then_remove_remote() {
        hard_link_then_remove(new_remote().await).await
    }

    async fn hard_link_then_remove_both(mut case: TestCase<impl Receiver>) {
        const EXPECTED: &[u8] = b"And the lives we've reclaimed";
        let name1 = OsStr::new("refugee_lyrics.txt");
        let name2 = OsStr::new("rise_against_lyrics.txt");
        let mnt_path = case.mnt_dir();
        let path1 = mnt_path.join(name1);
        let path2 = mnt_path.join(name2);
        write(&path1, EXPECTED).await.expect("write failed");

        hard_link(&path1, &path2).await.expect("hard_link failed");
        remove_file(&path1)
            .await
            .expect("remove_file on path1 failed");
        remove_file(&path2)
            .await
            .expect("remove_file on path2 failed");

        let expected = case.initial_contents();
        let actual = file_names(read_dir(&mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(expected, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn hard_link_then_remove_both_local() {
        hard_link_then_remove_both(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn hard_link_then_remove_both_remote() {
        hard_link_then_remove_both(new_remote().await).await
    }

    async fn set_mode_bits(mut case: TestCase<impl Receiver>) {
        const EXPECTED: u32 = libc::S_IFREG | 0o777;
        let file_path = case.mnt_dir().join("bagobits");
        write(&file_path, []).await.expect("write failed");
        let original = metadata(&file_path)
            .await
            .expect("metadata failed")
            .permissions()
            .mode();
        assert_ne!(EXPECTED, original);

        set_permissions(&file_path, Permissions::from_mode(EXPECTED))
            .await
            .expect("set_permissions failed");

        let actual = metadata(&file_path)
            .await
            .expect("metadata failed")
            .permissions()
            .mode();
        assert_eq!(EXPECTED, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn set_mode_bits_local() {
        set_mode_bits(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn set_mode_bits_remote() {
        set_mode_bits(new_remote().await).await
    }

    async fn create_directory(mut case: TestCase<impl Receiver>) {
        const EXPECTED: &str = "etc";
        let mnt_path = case.mnt_dir();
        let dir_path = mnt_path.join(EXPECTED);
        let mut expected = case.initial_contents();
        expected.push(OsStr::new(EXPECTED));

        create_dir(&dir_path).await.expect("create_dir failed");

        let actual = file_names(read_dir(mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(expected, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn create_directory_local() {
        create_directory(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn create_directory_remote() {
        create_directory(new_remote().await).await
    }

    async fn create_file_under_new_directory(mut case: TestCase<impl Receiver>) {
        const DIR_NAME: &str = "etc";
        const FILE_NAME: &str = "file";
        let mnt_path = case.mnt_dir();
        let dir_path = mnt_path.join(DIR_NAME);
        let file_path = dir_path.join(FILE_NAME);

        create_dir(&dir_path).await.expect("create_dir failed");
        write(&file_path, []).await.expect("write failed");

        let actual = file_names(read_dir(dir_path).await.expect("read_dir failed")).await;
        assert_eq!([FILE_NAME].as_slice(), actual.as_slice());

        case.stop().await;
    }

    #[tokio::test]
    async fn create_file_under_new_directory_local() {
        create_file_under_new_directory(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn create_file_under_new_directory_remote() {
        create_file_under_new_directory(new_remote().await).await
    }

    async fn create_then_remove_directory(mut case: TestCase<impl Receiver>) {
        const DIR_NAME: &str = "etc";
        let mnt_path = case.mnt_dir();
        let dir_path = mnt_path.join(DIR_NAME);

        create_dir(&dir_path).await.expect("create_dir failed");
        remove_dir(&dir_path).await.expect("remove_dir failed");

        let actual = file_names(read_dir(&mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(case.initial_contents(), actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn create_then_remote_directory_local() {
        create_then_remove_directory(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn create_then_remote_directory_remote() {
        create_then_remove_directory(new_remote().await).await
    }

    async fn read_only_dir_cant_create_subdir(mut case: TestCase<impl Receiver>) {
        const DIR_NAME: &str = "etc";
        let dir_path = case.mnt_dir().join(DIR_NAME);
        create_dir(&dir_path).await.expect("create_dir failed");
        set_permissions(&dir_path, Permissions::from_mode(libc::S_IFDIR | 0o444))
            .await
            .expect("set_permissions failed");

        let result = create_dir(dir_path.join("sub")).await;

        let err = result.err().expect("create_dir returned `Ok`");
        let os_err = err.raw_os_error().expect("raw_os_error was empty");
        assert_eq!(os_err, libc::EACCES);

        case.stop().await;
    }

    #[tokio::test]
    async fn read_only_dir_cant_create_subdir_local() {
        read_only_dir_cant_create_subdir(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn read_only_dir_cant_create_subdir_remote() {
        read_only_dir_cant_create_subdir(new_remote().await).await
    }

    async fn read_only_dir_cant_remove_subdir(mut case: TestCase<impl Receiver>) {
        const DIR_NAME: &str = "etc";
        let dir_path = case.mnt_dir().join(DIR_NAME);
        let sub_path = dir_path.join("sub");
        create_dir(&dir_path).await.expect("create_dir failed");
        create_dir(&sub_path).await.expect("create_dir failed");
        set_permissions(&dir_path, Permissions::from_mode(libc::S_IFDIR | 0o444))
            .await
            .expect("set_permissions failed");

        let result = remove_dir(&sub_path).await;

        let err = result.err().expect("remove_dir returned `Ok`");
        let os_err = err.raw_os_error().expect("raw_os_error was empty");
        assert_eq!(os_err, libc::EACCES);

        case.stop().await;
    }

    #[tokio::test]
    async fn read_only_dir_cant_remove_subdir_local() {
        read_only_dir_cant_remove_subdir(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn read_only_dir_cant_remove_subdir_remote() {
        read_only_dir_cant_remove_subdir(new_remote().await).await
    }

    async fn rename_file(mut case: TestCase<impl Receiver>) {
        const FILE_NAME: &str = "parabola.txt";
        const EXPECTED: &[u8] = b"We are eternal all this pain is an illusion";
        let src_path = case.mnt_dir().join(FILE_NAME);
        let dst_path = case.mnt_dir().join("parabola_lyrics.txt");

        write(&src_path, EXPECTED).await.unwrap();
        rename(&src_path, &dst_path).await.unwrap();

        let actual = read(&dst_path).await.unwrap();
        assert_eq!(EXPECTED, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn rename_file_local() {
        rename_file(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn rename_file_remote() {
        rename_file(new_remote().await).await
    }

    async fn write_read_with_file_struct(mut case: TestCase<impl Receiver>) {
        const FILE_NAME: &str = "big.dat";
        const LEN: usize = btlib::SECTOR_SZ_DEFAULT + 1;
        fn fill(buf: &mut Vec<u8>, value: u8) {
            buf.clear();
            buf.extend(std::iter::repeat(value).take(buf.capacity()));
        }

        let file_path = case.mnt_dir().join(FILE_NAME);
        let mut buf = vec![1u8; LEN];
        let mut file = OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .open(&file_path)
            .await
            .unwrap();

        file.write_all(&buf).await.unwrap();
        fill(&mut buf, 2);
        file.write_all(&buf).await.unwrap();
        file.rewind().await.unwrap();
        let mut actual = vec![0u8; LEN];
        file.read_exact(&mut actual).await.unwrap();

        fill(&mut buf, 1);
        assert_eq!(buf, actual);

        drop(file);
        case.stop().await;
    }

    #[tokio::test]
    async fn write_read_with_file_struct_local() {
        write_read_with_file_struct(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn write_read_with_file_struct_remote() {
        write_read_with_file_struct(new_remote().await).await
    }

    /// KMC: This test is currently not working, and I've not been able to figure out why, nor
    /// reproduce it at a lower layer of the stack.
    async fn read_more_than_whats_buffered(mut case: TestCase<impl Receiver>) {
        const FILE_NAME: &str = "big.dat";
        const SECT_SZ: usize = btlib::SECTOR_SZ_DEFAULT;
        const DIVISOR: usize = 8;
        const READ_SZ: usize = SECT_SZ / DIVISOR;

        let file_path = case.mnt_dir().join(FILE_NAME);
        let mut file = OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .open(&file_path)
            .await
            .unwrap();

        let mut buf = vec![1u8; 2 * SECT_SZ];
        file.write_all(&buf).await.unwrap();
        file.flush().await.unwrap();
        let mut file = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&file_path)
            .await
            .unwrap();
        file.seek(SeekFrom::Start(SECT_SZ as u64)).await.unwrap();
        let mut actual = vec![0u8; READ_SZ];
        file.read_exact(&mut actual).await.unwrap();

        buf.truncate(READ_SZ);
        assert!(buf == actual);

        case.stop().await;
    }

    //#[tokio::test]
    #[allow(dead_code)]
    async fn read_more_than_whats_buffered_local() {
        read_more_than_whats_buffered(new_local().await).await
    }

    //#[tokio::test(flavor = "multi_thread")]
    #[allow(dead_code)]
    async fn read_more_than_whats_buffered_remote() {
        read_more_than_whats_buffered(new_remote().await).await
    }
}