virtual_fs/

mount_fs.rs

//! A mount-topology filesystem that routes operations by path,
//! its not as simple as TmpFs. not currently used but was used by
//! the previously implementation of Deploy - now using TmpFs

use crate::*;

use std::{
    borrow::Cow,
    collections::{BTreeMap, BTreeSet},
    ffi::OsString,
    path::{Path, PathBuf},
    sync::{Arc, RwLock},
    time::{SystemTime, UNIX_EPOCH},
};

const MIN_METADATA_TIMESTAMP: u64 = 1_000_000_000; // 1 second in nano seconds

type DynFileSystem = Arc<dyn FileSystem + Send + Sync>;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ExactMountConflictMode {
    Fail,
    KeepExisting,
    ReplaceExisting,
}

#[derive(Debug, Clone)]
struct MountedFileSystem {
    fs: DynFileSystem,
    source_path: PathBuf,
}

#[derive(Debug, Default)]
struct MountNode {
    /// Creation timestamp in nanoseconds since the Unix epoch.
    /// Set once when the node is first inserted into the tree.
    created_at: u64,
    mount: Option<MountedFileSystem>,
    children: BTreeMap<OsString, MountNode>,
}

#[derive(Debug, Clone)]
struct ExactNode {
    path: PathBuf,
    fs: Option<DynFileSystem>,
    source_path: PathBuf,
    child_names: BTreeSet<OsString>,
    /// Timestamp reported for this node in nanoseconds since the Unix epoch.
    /// For synthetic non-mounted nodes, this may be generated at lookup time
    /// rather than representing an original creation event.
    created_at: u64,
}

impl ExactNode {
    fn has_children(&self) -> bool {
        !self.child_names.is_empty()
    }
}

#[derive(Debug, Clone)]
struct ResolvedMount {
    mount_path: PathBuf,
    delegated_path: PathBuf,
    fs: DynFileSystem,
}

#[derive(Debug, Clone)]
pub struct MountPoint {
    pub path: PathBuf,
    pub name: String,
    pub fs: Option<DynFileSystem>,
    pub children: Option<Arc<MountFileSystem>>,
}

#[derive(Debug, Clone)]
pub struct MountEntry {
    pub path: PathBuf,
    pub fs: DynFileSystem,
    pub source_path: PathBuf,
}

impl MountPoint {
    pub fn fs(&self) -> Option<&(dyn FileSystem + Send + Sync)> {
        self.fs.as_deref()
    }

    pub fn mount_point_ref(&self) -> MountPointRef<'_> {
        MountPointRef {
            path: self.path.clone(),
            name: self.name.clone(),
            fs: self.fs.as_deref(),
        }
    }
}

/// Allows different filesystems of different types
/// to be mounted at various mount points
#[derive(Debug)]
pub struct MountFileSystem {
    root: RwLock<MountNode>,
}

impl Default for MountFileSystem {
    fn default() -> Self {
        Self::new()
    }
}

impl MountFileSystem {
    pub fn new() -> Self {
        let ts = Self::now_nanos();
        Self {
            root: RwLock::new(MountNode {
                created_at: ts,
                ..MountNode::default()
            }),
        }
    }

    pub fn mount(
        &self,
        path: impl AsRef<Path>,
        fs: Arc<dyn FileSystem + Send + Sync>,
    ) -> Result<()> {
        self.mount_with_source(path, Path::new("/"), fs)
    }

    pub fn mount_with_source(
        &self,
        path: impl AsRef<Path>,
        source_path: impl AsRef<Path>,
        fs: Arc<dyn FileSystem + Send + Sync>,
    ) -> Result<()> {
        let path = self.prepare_path(path.as_ref())?;
        let source_path = Self::normalize_source_path(source_path.as_ref());
        let ts = Self::now_nanos();
        let mut root = self.root.write().unwrap();
        let node = Self::mount_node_mut(&mut root, &Self::path_components(&path), ts);

        if node.mount.is_some() {
            Err(FsError::AlreadyExists)
        } else {
            node.mount = Some(MountedFileSystem { fs, source_path });
            Ok(())
        }
    }

    pub fn filesystem_at(
        &self,
        path: impl AsRef<Path>,
    ) -> Option<Arc<dyn FileSystem + Send + Sync>> {
        self.exact_node(path.as_ref()).and_then(|node| node.fs)
    }

    pub fn clear(&mut self) {
        *self.root.write().unwrap() = MountNode {
            created_at: Self::now_nanos(),
            ..MountNode::default()
        };
    }

    fn prepare_path(&self, path: &Path) -> Result<PathBuf> {
        let mut normalized = PathBuf::new();

        for component in path.components() {
            match component {
                std::path::Component::RootDir | std::path::Component::CurDir => {}
                std::path::Component::ParentDir => {
                    if !normalized.pop() {
                        return Err(FsError::InvalidInput);
                    }
                }
                std::path::Component::Normal(part) => normalized.push(part),
                std::path::Component::Prefix(_) => return Err(FsError::InvalidInput),
            }
        }

        Ok(normalized)
    }

    fn path_components(path: &Path) -> Vec<OsString> {
        path.components()
            .map(|component| component.as_os_str().to_os_string())
            .collect()
    }

    fn absolute_path(components: &[OsString]) -> PathBuf {
        let mut path = PathBuf::from("/");
        for component in components {
            path.push(component);
        }
        path
    }

    fn normalize_source_path(path: &Path) -> PathBuf {
        let mut normalized = PathBuf::from("/");
        normalized.push(path.strip_prefix("/").unwrap_or(path));
        normalized
    }

    fn now_nanos() -> u64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_or(0, |d| d.as_nanos() as u64)
            .max(MIN_METADATA_TIMESTAMP)
    }

    fn directory_metadata_at(ts: u64) -> Metadata {
        Metadata {
            ft: FileType::new_dir(),
            accessed: ts,
            created: ts,
            modified: ts,
            len: 0,
        }
    }

    fn should_fallback_to_synthetic_dir(error: &FsError) -> bool {
        matches!(
            error,
            FsError::Unsupported | FsError::NotAFile | FsError::BaseNotDirectory
        )
    }

    fn synthetic_entry(name: OsString, base: &Path, ts: u64) -> DirEntry {
        DirEntry {
            path: base.join(PathBuf::from(name)),
            metadata: Ok(Self::directory_metadata_at(ts)),
        }
    }

    fn mounted(node: &MountNode) -> Option<MountedFileSystem> {
        node.mount.clone()
    }

    fn collect_mount_entries(node: &MountNode, path: &Path, entries: &mut Vec<MountEntry>) {
        if let Some(mount) = Self::mounted(node) {
            entries.push(MountEntry {
                path: path.to_path_buf(),
                fs: mount.fs,
                source_path: mount.source_path,
            });
        }

        for (child_name, child) in &node.children {
            let child_path = path.join(child_name);
            Self::collect_mount_entries(child, &child_path, entries);
        }
    }

    fn find_node<'a>(node: &'a MountNode, components: &[OsString]) -> Option<&'a MountNode> {
        let mut node = node;
        for component in components {
            node = node.children.get(component)?;
        }
        Some(node)
    }

    fn exact_node(&self, path: &Path) -> Option<ExactNode> {
        let path = self.prepare_path(path).ok()?;
        let components = Self::path_components(&path);
        let visible_path = Path::new("/").join(&path);
        let root = self.root.read().unwrap();
        let node = Self::find_node(&root, &components)?;
        let mounted = Self::mounted(node);

        Some(ExactNode {
            path: visible_path.clone(),
            fs: mounted.as_ref().map(|mount| mount.fs.clone()),
            created_at: node.created_at,
            source_path: mounted
                .map(|mount| mount.source_path)
                .unwrap_or_else(|| PathBuf::from("/")),
            child_names: node.children.keys().cloned().collect(),
        })
    }

    fn resolve_mount(&self, path: impl AsRef<Path>) -> Option<ResolvedMount> {
        let path = self.prepare_path(path.as_ref()).ok()?;
        let components = Self::path_components(&path);
        let root = self.root.read().unwrap();
        let mut node = &*root;
        let mut best = Self::mounted(node).map(|mount| ResolvedMount {
            mount_path: PathBuf::from("/"),
            delegated_path: mount.source_path.join(
                Self::absolute_path(&components)
                    .strip_prefix("/")
                    .unwrap_or(Path::new("")),
            ),
            fs: mount.fs,
        });

        for (index, component) in components.iter().enumerate() {
            let Some(child) = node.children.get(component) else {
                break;
            };
            node = child;

            if let Some(mount) = Self::mounted(node) {
                best = Some(ResolvedMount {
                    mount_path: Self::absolute_path(&components[..=index]),
                    delegated_path: mount.source_path.join(
                        Self::absolute_path(&components[index + 1..])
                            .strip_prefix("/")
                            .unwrap_or(Path::new("")),
                    ),
                    fs: mount.fs,
                });
            }
        }

        best
    }

    /// Creates, in the nearest enclosing mount, the directories that are only
    /// visible because a deeper mount sits below them (e.g. `/usr` for a mount
    /// at `/usr/local/ssl`). Without this, such a directory stats as a
    /// directory but nothing can be created inside it. Best-effort: errors are
    /// left for the caller's own operation to report.
    fn back_synthetic_dirs(&self, components: &[OsString]) {
        let mut pending = Vec::new();
        {
            let root = self.root.read().unwrap();
            let mut node = &*root;
            let mut mount = Self::mounted(node);
            let mut mount_depth = 0;

            for (index, component) in components.iter().enumerate() {
                let Some(child) = node.children.get(component) else {
                    break;
                };
                node = child;

                if let Some(child_mount) = Self::mounted(node) {
                    mount = Some(child_mount);
                    mount_depth = index + 1;
                    pending.clear();
                } else if let Some(mount) = &mount {
                    let relative = Self::absolute_path(&components[mount_depth..=index]);
                    let relative = relative.strip_prefix("/").unwrap_or(Path::new(""));
                    pending.push((mount.fs.clone(), mount.source_path.join(relative)));
                }
            }
        }

        for (fs, path) in pending {
            let _ = fs.create_dir(&path);
        }
    }

    fn back_synthetic_parents(&self, path: &Path) {
        let components = Self::path_components(path);
        if let Some((_, parents)) = components.split_last() {
            self.back_synthetic_dirs(parents);
        }
    }

    fn rebase_entries(entries: &mut ReadDir, source_prefix: &Path, target_prefix: &Path) {
        for entry in &mut entries.data {
            let suffix = entry.path.strip_prefix(source_prefix).unwrap_or_else(|_| {
                entry
                    .path
                    .strip_prefix(Path::new("/"))
                    .unwrap_or(&entry.path)
            });
            entry.path = target_prefix.join(suffix);
        }
    }

    fn read_dir_from_exact_node(&self, node: &ExactNode) -> Result<ReadDir> {
        let mut entries = Vec::new();

        let backing = if let Some(fs) = &node.fs {
            Some((
                fs.read_dir(&node.source_path),
                Cow::Borrowed(node.source_path.as_path()),
            ))
        } else {
            self.resolve_mount(&node.path).map(|resolved| {
                (
                    resolved.fs.read_dir(&resolved.delegated_path),
                    Cow::Owned(resolved.delegated_path),
                )
            })
        };

        if let Some((base_entries, source_path)) = backing {
            match base_entries {
                Ok(mut base_entries) => {
                    Self::rebase_entries(&mut base_entries, &source_path, &node.path);
                    entries.extend(base_entries.data.into_iter().filter(|entry| {
                        entry
                            .path
                            .file_name()
                            .map(|name| !node.child_names.contains(name))
                            .unwrap_or(true)
                    }));
                }
                Err(FsError::EntryNotFound) if node.has_children() => {}
                Err(error)
                    if node.has_children() && Self::should_fallback_to_synthetic_dir(&error) => {}
                Err(error) => return Err(error),
            }
        }

        entries.extend(
            node.child_names
                .iter()
                .cloned()
                .map(|name| Self::synthetic_entry(name, &node.path, node.created_at)),
        );

        Ok(ReadDir::new(entries))
    }

    fn mount_node_mut<'a>(
        node: &'a mut MountNode,
        components: &[OsString],
        ts: u64,
    ) -> &'a mut MountNode {
        let mut node = node;
        for component in components {
            node = node
                .children
                .entry(component.clone())
                .or_insert_with(|| MountNode {
                    created_at: ts,
                    ..MountNode::default()
                });
        }

        node
    }

    fn clear_descendants(node: &mut MountNode) {
        node.children.clear();
    }

    /// Overwrite the mount at `path`.
    pub fn set_mount(
        &self,
        path: impl AsRef<Path>,
        fs: Arc<dyn FileSystem + Send + Sync>,
    ) -> Result<()> {
        let path = self.prepare_path(path.as_ref())?;
        let ts = Self::now_nanos();
        let mut root = self.root.write().unwrap();
        let node = Self::mount_node_mut(&mut root, &Self::path_components(&path), ts);
        node.mount = Some(MountedFileSystem {
            fs,
            source_path: PathBuf::from("/"),
        });
        Ok(())
    }

    pub fn add_mount_entries_with_mode(
        &self,
        entries: impl IntoIterator<Item = MountEntry>,
        conflict_mode: ExactMountConflictMode,
    ) -> Result<()> {
        let mut skipped_subtrees = Vec::<PathBuf>::new();

        for entry in entries {
            if skipped_subtrees
                .iter()
                .any(|prefix| entry.path.starts_with(prefix))
            {
                continue;
            }

            let exact_conflict = self.filesystem_at(&entry.path).is_some();
            if exact_conflict {
                match conflict_mode {
                    ExactMountConflictMode::Fail => return Err(FsError::AlreadyExists),
                    ExactMountConflictMode::KeepExisting => {
                        skipped_subtrees.push(entry.path);
                        continue;
                    }
                    ExactMountConflictMode::ReplaceExisting => {
                        let ts = Self::now_nanos();
                        let mut root = self.root.write().unwrap();
                        let node = Self::mount_node_mut(
                            &mut root,
                            &Self::path_components(&self.prepare_path(&entry.path)?),
                            ts,
                        );
                        Self::clear_descendants(node);
                        node.mount = Some(MountedFileSystem {
                            fs: entry.fs,
                            source_path: entry.source_path,
                        });
                        continue;
                    }
                }
            }

            self.mount_with_source(&entry.path, &entry.source_path, entry.fs)?;
        }

        Ok(())
    }
    pub fn mount_entries(&self) -> Vec<MountEntry> {
        let mut entries = Vec::new();
        let root = self.root.read().unwrap();
        Self::collect_mount_entries(&root, Path::new("/"), &mut entries);
        entries
    }
}

impl FileSystem for MountFileSystem {
    fn readlink(&self, path: &Path) -> Result<PathBuf> {
        let path = self.prepare_path(path)?;

        if path.as_os_str().is_empty() {
            Err(FsError::NotAFile)
        } else {
            if let Some(node) = self.exact_node(&path)
                && node.fs.is_none()
            {
                return Err(FsError::EntryNotFound);
            }

            match self.resolve_mount(path) {
                Some(resolved) => resolved.fs.readlink(&resolved.delegated_path),
                None => Err(FsError::EntryNotFound),
            }
        }
    }

    fn read_dir(&self, path: &Path) -> Result<ReadDir> {
        let path = self.prepare_path(path)?;

        if let Some(node) = self.exact_node(&path) {
            return self.read_dir_from_exact_node(&node);
        }

        match self.resolve_mount(path.clone()) {
            Some(resolved) => {
                let mut entries = resolved.fs.read_dir(&resolved.delegated_path)?;
                Self::rebase_entries(
                    &mut entries,
                    &resolved.delegated_path,
                    &Path::new("/").join(&path),
                );
                Ok(entries)
            }
            None => Err(FsError::EntryNotFound),
        }
    }

    fn create_dir(&self, path: &Path) -> Result<()> {
        let path = self.prepare_path(path)?;

        if path.as_os_str().is_empty() {
            return Ok(());
        }

        if let Some(node) = self.exact_node(&path) {
            return if let Some(fs) = node.fs {
                let result = fs.create_dir(Path::new("/"));

                match result {
                    Ok(()) | Err(FsError::AlreadyExists) => Ok(()),
                    Err(error) if Self::should_fallback_to_synthetic_dir(&error) => Ok(()),
                    Err(error) => Err(error),
                }
            } else {
                self.back_synthetic_dirs(&Self::path_components(&path));
                Ok(())
            };
        }

        self.back_synthetic_parents(&path);
        match self.resolve_mount(path) {
            Some(resolved) => {
                let result = resolved.fs.create_dir(&resolved.delegated_path);

                if let Err(error) = result
                    && error == FsError::AlreadyExists
                {
                    return Ok(());
                }

                result
            }
            None => Err(FsError::EntryNotFound),
        }
    }

    fn create_symlink(&self, source: &Path, target: &Path) -> Result<()> {
        let target = self.prepare_path(target)?;

        if target.as_os_str().is_empty() {
            return Err(FsError::AlreadyExists);
        }

        if self.exact_node(&target).is_some() {
            return Err(FsError::AlreadyExists);
        }

        match self.resolve_mount(target) {
            Some(resolved) => resolved.fs.create_symlink(source, &resolved.delegated_path),
            None => Err(FsError::EntryNotFound),
        }
    }

    fn hard_link(&self, source: &Path, target: &Path) -> Result<()> {
        let source = self.prepare_path(source)?;
        let target = self.prepare_path(target)?;

        if source.as_os_str().is_empty() {
            return Err(FsError::PermissionDenied);
        }

        if target.as_os_str().is_empty() {
            return Err(FsError::AlreadyExists);
        }

        if let Some(node) = self.exact_node(&target)
            && (node.fs.is_some() || node.has_children())
        {
            return Err(FsError::AlreadyExists);
        }

        match (self.resolve_mount(source), self.resolve_mount(target)) {
            (Some(source_mount), Some(target_mount))
                if source_mount.mount_path == target_mount.mount_path =>
            {
                source_mount
                    .fs
                    .hard_link(&source_mount.delegated_path, &target_mount.delegated_path)
            }
            (Some(_), Some(_)) => Err(FsError::Unsupported),
            _ => Err(FsError::EntryNotFound),
        }
    }

    fn remove_dir(&self, path: &Path) -> Result<()> {
        let path = self.prepare_path(path)?;

        if path.as_os_str().is_empty() {
            return Err(FsError::PermissionDenied);
        }

        if let Some(node) = self.exact_node(&path) {
            return if node.fs.is_some() || node.has_children() {
                Err(FsError::PermissionDenied)
            } else {
                Err(FsError::EntryNotFound)
            };
        }

        match self.resolve_mount(path) {
            Some(resolved) => resolved.fs.remove_dir(&resolved.delegated_path),
            None => Err(FsError::EntryNotFound),
        }
    }

    fn rename<'a>(&'a self, from: &'a Path, to: &'a Path) -> BoxFuture<'a, Result<()>> {
        Box::pin(async move {
            let from = self.prepare_path(from)?;
            let to = self.prepare_path(to)?;

            if from.as_os_str().is_empty() {
                return Err(FsError::PermissionDenied);
            }

            if let Some(node) = self.exact_node(&from)
                && (node.fs.is_some() || node.has_children())
            {
                return Err(FsError::PermissionDenied);
            }

            if let Some(node) = self.exact_node(&to)
                && (node.fs.is_some() || node.has_children())
            {
                return Err(FsError::PermissionDenied);
            }

            match (self.resolve_mount(from), self.resolve_mount(to)) {
                (Some(from_mount), Some(to_mount))
                    if from_mount.mount_path == to_mount.mount_path =>
                {
                    from_mount
                        .fs
                        .rename(&from_mount.delegated_path, &to_mount.delegated_path)
                        .await
                }
                (Some(from_mount), Some(to_mount)) => {
                    ops::move_across_filesystems(
                        from_mount.fs.as_ref(),
                        to_mount.fs.as_ref(),
                        &from_mount.delegated_path,
                        &to_mount.delegated_path,
                    )
                    .await
                }
                _ => Err(FsError::EntryNotFound),
            }
        })
    }

    fn metadata(&self, path: &Path) -> Result<Metadata> {
        let path = self.prepare_path(path)?;

        if let Some(node) = self.exact_node(&path) {
            return if let Some(fs) = node.fs {
                fs.metadata(&node.source_path).or_else(|error| {
                    if Self::should_fallback_to_synthetic_dir(&error) {
                        Ok(Self::directory_metadata_at(node.created_at))
                    } else {
                        Err(error)
                    }
                })
            } else if node.has_children() {
                Ok(Self::directory_metadata_at(node.created_at))
            } else {
                Err(FsError::EntryNotFound)
            };
        }

        match self.resolve_mount(path) {
            Some(resolved) => resolved.fs.metadata(&resolved.delegated_path),
            None => Err(FsError::EntryNotFound),
        }
    }

    fn symlink_metadata(&self, path: &Path) -> Result<Metadata> {
        let path = self.prepare_path(path)?;

        if let Some(node) = self.exact_node(&path) {
            return if let Some(fs) = node.fs {
                fs.symlink_metadata(&node.source_path).or_else(|error| {
                    if Self::should_fallback_to_synthetic_dir(&error) {
                        Ok(Self::directory_metadata_at(node.created_at))
                    } else {
                        Err(error)
                    }
                })
            } else if node.has_children() {
                Ok(Self::directory_metadata_at(node.created_at))
            } else {
                Err(FsError::EntryNotFound)
            };
        }

        match self.resolve_mount(path) {
            Some(resolved) => resolved.fs.symlink_metadata(&resolved.delegated_path),
            None => Err(FsError::EntryNotFound),
        }
    }

    fn remove_file(&self, path: &Path) -> Result<()> {
        let path = self.prepare_path(path)?;

        if path.as_os_str().is_empty() {
            return Err(FsError::NotAFile);
        }

        if let Some(node) = self.exact_node(&path) {
            return if node.fs.is_some() || node.has_children() {
                Err(FsError::PermissionDenied)
            } else {
                Err(FsError::EntryNotFound)
            };
        }

        match self.resolve_mount(path) {
            Some(resolved) => resolved.fs.remove_file(&resolved.delegated_path),
            None => Err(FsError::EntryNotFound),
        }
    }

    fn new_open_options(&self) -> OpenOptions<'_> {
        OpenOptions::new(self)
    }
}

#[derive(Debug)]
pub struct MountPointRef<'a> {
    pub path: PathBuf,
    pub name: String,
    pub fs: Option<&'a (dyn FileSystem + Send + Sync)>,
}

impl FileOpener for MountFileSystem {
    fn open(
        &self,
        path: &Path,
        conf: &OpenOptionsConfig,
    ) -> Result<Box<dyn VirtualFile + Send + Sync>> {
        let path = self.prepare_path(path)?;

        if path.as_os_str().is_empty() {
            return Err(FsError::NotAFile);
        }

        if let Some(node) = self.exact_node(&path)
            && node.fs.is_none()
        {
            return Err(FsError::NotAFile);
        }

        if conf.create || conf.create_new {
            self.back_synthetic_parents(&path);
        }
        match self.resolve_mount(path) {
            Some(resolved) => resolved
                .fs
                .new_open_options()
                .options(conf.clone())
                .open(resolved.delegated_path),
            None => Err(FsError::EntryNotFound),
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{
        collections::HashSet,
        path::{Path, PathBuf},
        sync::Arc,
    };

    use tokio::io::AsyncWriteExt;

    use crate::{FileSystem as FileSystemTrait, FsError, MountFileSystem, TmpFileSystem, mem_fs};

    use super::{FileOpener, OpenOptionsConfig};

    #[derive(Debug, Clone, Default)]
    struct MountlessFileSystem {
        inner: mem_fs::FileSystem,
    }

    #[derive(Debug, Clone, Default)]
    struct RootOpaqueFileSystem {
        inner: mem_fs::FileSystem,
    }

    #[derive(Debug, Clone, Default)]
    struct RootPermissionDeniedFileSystem;

    impl FileSystemTrait for MountlessFileSystem {
        fn readlink(&self, path: &Path) -> crate::Result<PathBuf> {
            self.inner.readlink(path)
        }

        fn read_dir(&self, path: &Path) -> crate::Result<crate::ReadDir> {
            self.inner.read_dir(path)
        }

        fn create_dir(&self, path: &Path) -> crate::Result<()> {
            self.inner.create_dir(path)
        }

        fn remove_dir(&self, path: &Path) -> crate::Result<()> {
            self.inner.remove_dir(path)
        }

        fn rename<'a>(
            &'a self,
            from: &'a Path,
            to: &'a Path,
        ) -> futures::future::BoxFuture<'a, crate::Result<()>> {
            Box::pin(async move { self.inner.rename(from, to).await })
        }

        fn metadata(&self, path: &Path) -> crate::Result<crate::Metadata> {
            self.inner.metadata(path)
        }

        fn symlink_metadata(&self, path: &Path) -> crate::Result<crate::Metadata> {
            self.inner.symlink_metadata(path)
        }

        fn remove_file(&self, path: &Path) -> crate::Result<()> {
            self.inner.remove_file(path)
        }

        fn new_open_options(&self) -> crate::OpenOptions<'_> {
            self.inner.new_open_options()
        }
    }

    impl FileOpener for MountlessFileSystem {
        fn open(
            &self,
            path: &Path,
            conf: &OpenOptionsConfig,
        ) -> crate::Result<Box<dyn crate::VirtualFile + Send + Sync>> {
            self.inner
                .new_open_options()
                .options(conf.clone())
                .open(path)
        }
    }

    impl FileSystemTrait for RootOpaqueFileSystem {
        fn readlink(&self, path: &Path) -> crate::Result<PathBuf> {
            self.inner.readlink(path)
        }

        fn read_dir(&self, path: &Path) -> crate::Result<crate::ReadDir> {
            if path == Path::new("/") {
                Err(FsError::Unsupported)
            } else {
                self.inner.read_dir(path)
            }
        }

        fn create_dir(&self, path: &Path) -> crate::Result<()> {
            self.inner.create_dir(path)
        }

        fn remove_dir(&self, path: &Path) -> crate::Result<()> {
            self.inner.remove_dir(path)
        }

        fn rename<'a>(
            &'a self,
            from: &'a Path,
            to: &'a Path,
        ) -> futures::future::BoxFuture<'a, crate::Result<()>> {
            Box::pin(async move { self.inner.rename(from, to).await })
        }

        fn metadata(&self, path: &Path) -> crate::Result<crate::Metadata> {
            if path == Path::new("/") {
                Err(FsError::Unsupported)
            } else {
                self.inner.metadata(path)
            }
        }

        fn symlink_metadata(&self, path: &Path) -> crate::Result<crate::Metadata> {
            if path == Path::new("/") {
                Err(FsError::Unsupported)
            } else {
                self.inner.symlink_metadata(path)
            }
        }

        fn remove_file(&self, path: &Path) -> crate::Result<()> {
            self.inner.remove_file(path)
        }

        fn new_open_options(&self) -> crate::OpenOptions<'_> {
            self.inner.new_open_options()
        }
    }

    impl FileOpener for RootOpaqueFileSystem {
        fn open(
            &self,
            path: &Path,
            conf: &OpenOptionsConfig,
        ) -> crate::Result<Box<dyn crate::VirtualFile + Send + Sync>> {
            self.inner
                .new_open_options()
                .options(conf.clone())
                .open(path)
        }
    }

    impl FileSystemTrait for RootPermissionDeniedFileSystem {
        fn readlink(&self, _path: &Path) -> crate::Result<PathBuf> {
            Err(FsError::PermissionDenied)
        }

        fn read_dir(&self, _path: &Path) -> crate::Result<crate::ReadDir> {
            Err(FsError::PermissionDenied)
        }

        fn create_dir(&self, _path: &Path) -> crate::Result<()> {
            Err(FsError::PermissionDenied)
        }

        fn remove_dir(&self, _path: &Path) -> crate::Result<()> {
            Err(FsError::PermissionDenied)
        }

        fn rename<'a>(
            &'a self,
            _from: &'a Path,
            _to: &'a Path,
        ) -> futures::future::BoxFuture<'a, crate::Result<()>> {
            Box::pin(async { Err(FsError::PermissionDenied) })
        }

        fn metadata(&self, _path: &Path) -> crate::Result<crate::Metadata> {
            Err(FsError::PermissionDenied)
        }

        fn symlink_metadata(&self, _path: &Path) -> crate::Result<crate::Metadata> {
            Err(FsError::PermissionDenied)
        }

        fn remove_file(&self, _path: &Path) -> crate::Result<()> {
            Err(FsError::PermissionDenied)
        }

        fn new_open_options(&self) -> crate::OpenOptions<'_> {
            crate::OpenOptions::new(self)
        }
    }

    impl FileOpener for RootPermissionDeniedFileSystem {
        fn open(
            &self,
            _path: &Path,
            _conf: &OpenOptionsConfig,
        ) -> crate::Result<Box<dyn crate::VirtualFile + Send + Sync>> {
            Err(FsError::PermissionDenied)
        }
    }

    fn gen_filesystem() -> MountFileSystem {
        let union = MountFileSystem::new();
        let a = mem_fs::FileSystem::default();
        let b = mem_fs::FileSystem::default();
        let c = mem_fs::FileSystem::default();
        let d = mem_fs::FileSystem::default();
        let e = mem_fs::FileSystem::default();
        let f = mem_fs::FileSystem::default();
        let g = mem_fs::FileSystem::default();
        let h = mem_fs::FileSystem::default();

        union
            .mount(PathBuf::from("/test_new_filesystem").as_path(), Arc::new(a))
            .unwrap();
        union
            .mount(PathBuf::from("/test_create_dir").as_path(), Arc::new(b))
            .unwrap();
        union
            .mount(PathBuf::from("/test_remove_dir").as_path(), Arc::new(c))
            .unwrap();
        union
            .mount(PathBuf::from("/test_rename").as_path(), Arc::new(d))
            .unwrap();
        union
            .mount(PathBuf::from("/test_metadata").as_path(), Arc::new(e))
            .unwrap();
        union
            .mount(PathBuf::from("/test_remove_file").as_path(), Arc::new(f))
            .unwrap();
        union
            .mount(PathBuf::from("/test_readdir").as_path(), Arc::new(g))
            .unwrap();
        union
            .mount(PathBuf::from("/test_canonicalize").as_path(), Arc::new(h))
            .unwrap();

        union
    }

    fn gen_nested_filesystem() -> MountFileSystem {
        let union = MountFileSystem::new();
        let a = mem_fs::FileSystem::default();
        a.open(
            &PathBuf::from("/data-a.txt"),
            &OpenOptionsConfig {
                read: true,
                write: true,
                create_new: false,
                create: true,
                append: false,
                truncate: false,
            },
        )
        .unwrap();
        let b = mem_fs::FileSystem::default();
        b.open(
            &PathBuf::from("/data-b.txt"),
            &OpenOptionsConfig {
                read: true,
                write: true,
                create_new: false,
                create: true,
                append: false,
                truncate: false,
            },
        )
        .unwrap();

        union
            .mount(PathBuf::from("/app/a").as_path(), Arc::new(a))
            .unwrap();
        union
            .mount(PathBuf::from("/app/b").as_path(), Arc::new(b))
            .unwrap();

        union
    }

    #[tokio::test]
    async fn test_nested_read_dir() {
        let fs = gen_nested_filesystem();

        let root_contents: Vec<PathBuf> = fs
            .read_dir(&PathBuf::from("/"))
            .unwrap()
            .map(|e| e.unwrap().path.clone())
            .collect();
        assert_eq!(root_contents, vec![PathBuf::from("/app")]);

        let app_contents: HashSet<PathBuf> = fs
            .read_dir(&PathBuf::from("/app"))
            .unwrap()
            .map(|e| e.unwrap().path)
            .collect();
        assert_eq!(
            app_contents,
            HashSet::from_iter([PathBuf::from("/app/a"), PathBuf::from("/app/b")].into_iter())
        );

        let a_contents: Vec<PathBuf> = fs
            .read_dir(&PathBuf::from("/app/a"))
            .unwrap()
            .map(|e| e.unwrap().path.clone())
            .collect();
        assert_eq!(a_contents, vec![PathBuf::from("/app/a/data-a.txt")]);

        let b_contents: Vec<PathBuf> = fs
            .read_dir(&PathBuf::from("/app/b"))
            .unwrap()
            .map(|e| e.unwrap().path)
            .collect();
        assert_eq!(b_contents, vec![PathBuf::from("/app/b/data-b.txt")]);
    }

    #[tokio::test]
    async fn test_nested_metadata() {
        let fs = gen_nested_filesystem();

        assert!(fs.metadata(&PathBuf::from("/")).is_ok());
        assert!(fs.metadata(&PathBuf::from("/app")).is_ok());
        assert!(fs.metadata(&PathBuf::from("/app/a")).is_ok());
        assert!(fs.metadata(&PathBuf::from("/app/b")).is_ok());
        assert!(fs.metadata(&PathBuf::from("/app/a/data-a.txt")).is_ok());
        assert!(fs.metadata(&PathBuf::from("/app/b/data-b.txt")).is_ok());
    }

    #[tokio::test]
    async fn test_nested_symlink_metadata() {
        let fs = gen_nested_filesystem();

        assert!(fs.symlink_metadata(&PathBuf::from("/")).is_ok());
        assert!(fs.symlink_metadata(&PathBuf::from("/app")).is_ok());
        assert!(fs.symlink_metadata(&PathBuf::from("/app/a")).is_ok());
        assert!(fs.symlink_metadata(&PathBuf::from("/app/b")).is_ok());
        assert!(
            fs.symlink_metadata(&PathBuf::from("/app/a/data-a.txt"))
                .is_ok()
        );
        assert!(
            fs.symlink_metadata(&PathBuf::from("/app/b/data-b.txt"))
                .is_ok()
        );
    }

    #[tokio::test]
    async fn test_import_mounts_preserves_nested_root_mounts() {
        let primary = MountFileSystem::new();
        let openssl = mem_fs::FileSystem::default();
        openssl.create_dir(Path::new("/certs")).unwrap();
        openssl
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/certs/ca.pem"))
            .unwrap();
        primary
            .mount(Path::new("/openssl"), Arc::new(openssl))
            .unwrap();

        let injected = MountFileSystem::new();
        let app = mem_fs::FileSystem::default();
        app.new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/index.php"))
            .unwrap();
        injected.mount(Path::new("/app"), Arc::new(app)).unwrap();

        let assets = mem_fs::FileSystem::default();
        assets.create_dir(Path::new("/css")).unwrap();
        assets
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/css/site.css"))
            .unwrap();
        injected
            .mount(Path::new("/opt/assets"), Arc::new(assets))
            .unwrap();

        primary
            .add_mount_entries_with_mode(
                injected.mount_entries(),
                super::ExactMountConflictMode::Fail,
            )
            .unwrap();

        let root_contents = read_dir_names(&primary, "/");
        assert!(root_contents.contains(&"app".to_string()));
        assert!(root_contents.contains(&"opt".to_string()));
        assert!(root_contents.contains(&"openssl".to_string()));
        assert!(primary.metadata(Path::new("/app/index.php")).is_ok());
        assert!(
            primary
                .metadata(Path::new("/opt/assets/css/site.css"))
                .is_ok()
        );
        assert!(primary.metadata(Path::new("/openssl/certs/ca.pem")).is_ok());
    }

    #[tokio::test]
    async fn test_nested_mount_under_non_mountable_leaf_is_supported() {
        let fs = MountFileSystem::new();

        let top = MountlessFileSystem::default();
        top.create_dir(Path::new("/bin")).unwrap();
        top.new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/bin/tool"))
            .unwrap();

        let nested = mem_fs::FileSystem::default();
        nested.create_dir(Path::new("/css")).unwrap();
        nested
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/css/site.css"))
            .unwrap();

        fs.mount(Path::new("/opt"), Arc::new(top)).unwrap();
        fs.mount(Path::new("/opt/assets"), Arc::new(nested))
            .unwrap();

        assert!(fs.metadata(Path::new("/opt/bin/tool")).is_ok());
        assert!(fs.metadata(Path::new("/opt/assets/css/site.css")).is_ok());
    }

    #[tokio::test]
    async fn test_normalized_paths_still_route_to_deepest_mount() {
        let fs = MountFileSystem::new();

        let top = MountlessFileSystem::default();
        top.create_dir(Path::new("/bin")).unwrap();
        top.new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/bin/tool"))
            .unwrap();

        let nested = mem_fs::FileSystem::default();
        nested.create_dir(Path::new("/css")).unwrap();
        nested
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/css/site.css"))
            .unwrap();

        fs.mount(Path::new("/opt"), Arc::new(top)).unwrap();
        fs.mount(Path::new("/opt/assets"), Arc::new(nested))
            .unwrap();

        assert!(
            fs.metadata(Path::new("/opt/./assets/../assets/css/site.css"))
                .unwrap()
                .is_file()
        );
    }

    #[tokio::test]
    async fn test_invalid_above_root_path_is_rejected() {
        let fs = MountFileSystem::new();
        fs.mount(Path::new("/"), Arc::new(mem_fs::FileSystem::default()))
            .unwrap();

        assert_eq!(fs.metadata(Path::new("../foo")), Err(FsError::InvalidInput));
    }

    #[tokio::test]
    async fn test_exact_mount_metadata_falls_back_to_synthetic_directory() {
        let fs = MountFileSystem::new();
        fs.mount(
            Path::new("/opaque"),
            Arc::new(RootOpaqueFileSystem::default()),
        )
        .unwrap();

        let meta1 = fs.metadata(Path::new("/opaque")).unwrap();
        let sym1 = fs.symlink_metadata(Path::new("/opaque")).unwrap();
        assert!(meta1.is_dir());
        assert!(sym1.is_dir());

        // Timestamps must be non-zero (regression guard against the old `0` placeholders).
        assert!(meta1.created > 0, "created timestamp must be non-zero");
        assert!(meta1.modified > 0, "modified timestamp must be non-zero");
        assert!(meta1.accessed > 0, "accessed timestamp must be non-zero");

        // Repeated calls must return the same stable timestamps (not re-sampled each time).
        let meta2 = fs.metadata(Path::new("/opaque")).unwrap();
        assert_eq!(meta1.created, meta2.created, "created must be stable");
        assert_eq!(meta1.modified, meta2.modified, "modified must be stable");
        assert_eq!(meta1.accessed, meta2.accessed, "accessed must be stable");

        assert_eq!(fs.create_dir(Path::new("/opaque")), Ok(()));
    }

    #[tokio::test]
    async fn test_exact_mount_read_dir_falls_back_to_child_mounts_when_root_is_unlistable() {
        let fs = MountFileSystem::new();
        fs.mount(
            Path::new("/opaque"),
            Arc::new(RootOpaqueFileSystem::default()),
        )
        .unwrap();
        fs.mount(
            Path::new("/opaque/assets"),
            Arc::new(mem_fs::FileSystem::default()),
        )
        .unwrap();

        assert_eq!(read_dir_names(&fs, "/opaque"), vec!["assets".to_string()]);
    }

    #[tokio::test]
    async fn test_exact_mount_fallback_does_not_mask_permission_denied() {
        let fs = MountFileSystem::new();
        fs.mount(
            Path::new("/denied"),
            Arc::new(RootPermissionDeniedFileSystem),
        )
        .unwrap();
        fs.mount(
            Path::new("/denied/assets"),
            Arc::new(mem_fs::FileSystem::default()),
        )
        .unwrap();

        assert_eq!(
            fs.metadata(Path::new("/denied")),
            Err(FsError::PermissionDenied)
        );
        assert_eq!(
            fs.symlink_metadata(Path::new("/denied")),
            Err(FsError::PermissionDenied)
        );
        assert_eq!(
            fs.read_dir(Path::new("/denied")).map(|_| ()),
            Err(FsError::PermissionDenied)
        );
        assert_eq!(
            fs.create_dir(Path::new("/denied")),
            Err(FsError::PermissionDenied)
        );
    }

    #[tokio::test]
    async fn test_keep_existing_conflict_skips_the_other_subtree() {
        let primary = MountFileSystem::new();
        let user_mount = mem_fs::FileSystem::default();
        user_mount
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/user.txt"))
            .unwrap();
        primary
            .mount(Path::new("/python"), Arc::new(user_mount))
            .unwrap();

        let injected = MountFileSystem::new();
        let package_mount = mem_fs::FileSystem::default();
        package_mount
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/pkg.txt"))
            .unwrap();
        injected
            .mount(Path::new("/python"), Arc::new(package_mount))
            .unwrap();

        let package_child = mem_fs::FileSystem::default();
        package_child
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/child.txt"))
            .unwrap();
        injected
            .mount(Path::new("/python/lib"), Arc::new(package_child))
            .unwrap();

        primary
            .add_mount_entries_with_mode(
                injected.mount_entries(),
                super::ExactMountConflictMode::KeepExisting,
            )
            .unwrap();

        assert!(
            primary
                .metadata(Path::new("/python/user.txt"))
                .unwrap()
                .is_file()
        );
        assert_eq!(
            primary.metadata(Path::new("/python/pkg.txt")),
            Err(FsError::EntryNotFound)
        );
        assert_eq!(
            primary.metadata(Path::new("/python/lib/child.txt")),
            Err(FsError::EntryNotFound)
        );
    }

    #[tokio::test]
    async fn test_replace_existing_conflict_replaces_the_whole_subtree() {
        let primary = MountFileSystem::new();
        let user_mount = mem_fs::FileSystem::default();
        user_mount
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/user.txt"))
            .unwrap();
        let user_child = mem_fs::FileSystem::default();
        user_child
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/user-child.txt"))
            .unwrap();
        primary
            .mount(Path::new("/python"), Arc::new(user_mount))
            .unwrap();
        primary
            .mount(Path::new("/python/lib"), Arc::new(user_child))
            .unwrap();

        let injected = MountFileSystem::new();
        let package_mount = mem_fs::FileSystem::default();
        package_mount
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/pkg.txt"))
            .unwrap();
        let package_child = mem_fs::FileSystem::default();
        package_child
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/pkg-child.txt"))
            .unwrap();
        injected
            .mount(Path::new("/python"), Arc::new(package_mount))
            .unwrap();
        injected
            .mount(Path::new("/python/lib"), Arc::new(package_child))
            .unwrap();

        primary
            .add_mount_entries_with_mode(
                injected.mount_entries(),
                super::ExactMountConflictMode::ReplaceExisting,
            )
            .unwrap();

        assert_eq!(
            primary.metadata(Path::new("/python/user.txt")),
            Err(FsError::EntryNotFound)
        );
        assert_eq!(
            primary.metadata(Path::new("/python/lib/user-child.txt")),
            Err(FsError::EntryNotFound)
        );
        assert!(
            primary
                .metadata(Path::new("/python/pkg.txt"))
                .unwrap()
                .is_file()
        );
        assert!(
            primary
                .metadata(Path::new("/python/lib/pkg-child.txt"))
                .unwrap()
                .is_file()
        );
    }

    #[tokio::test]
    async fn test_exact_mountpoints_reject_destructive_mutation() {
        let fs = MountFileSystem::new();
        let mounted = mem_fs::FileSystem::default();
        mounted.create_dir(Path::new("/dir")).unwrap();
        mounted
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/file.txt"))
            .unwrap();

        fs.mount(Path::new("/mounted"), Arc::new(mounted)).unwrap();

        assert_eq!(
            fs.remove_dir(Path::new("/mounted")),
            Err(FsError::PermissionDenied)
        );
        assert_eq!(
            fs.remove_file(Path::new("/mounted")),
            Err(FsError::PermissionDenied)
        );
        assert_eq!(
            fs.rename(Path::new("/mounted"), Path::new("/other")).await,
            Err(FsError::PermissionDenied)
        );
        assert_eq!(
            fs.rename(Path::new("/mounted/file.txt"), Path::new("/mounted"))
                .await,
            Err(FsError::PermissionDenied)
        );
    }

    #[tokio::test]
    async fn test_parent_read_dir_merges_leaf_entries_with_child_mounts() {
        let fs = MountFileSystem::new();

        let top = MountlessFileSystem::default();
        top.create_dir(Path::new("/bin")).unwrap();
        top.new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/bin/tool"))
            .unwrap();

        let nested = mem_fs::FileSystem::default();
        nested.create_dir(Path::new("/css")).unwrap();
        nested
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/css/site.css"))
            .unwrap();

        fs.mount(Path::new("/opt"), Arc::new(top)).unwrap();
        fs.mount(Path::new("/opt/assets"), Arc::new(nested))
            .unwrap();

        let opt_contents = read_dir_names(&fs, "/opt");
        assert!(opt_contents.contains(&"bin".to_string()));
        assert!(opt_contents.contains(&"assets".to_string()));
    }

    #[tokio::test]
    async fn test_child_mount_shadows_same_named_parent_entry() {
        let fs = MountFileSystem::new();

        let top = MountlessFileSystem::default();
        top.new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/assets"))
            .unwrap();

        let nested = mem_fs::FileSystem::default();
        nested.create_dir(Path::new("/css")).unwrap();
        nested
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/css/site.css"))
            .unwrap();

        fs.mount(Path::new("/opt"), Arc::new(top)).unwrap();
        fs.mount(Path::new("/opt/assets"), Arc::new(nested))
            .unwrap();

        assert!(fs.metadata(Path::new("/opt/assets")).unwrap().is_dir());
        assert_eq!(
            read_dir_names(&fs, "/opt")
                .into_iter()
                .filter(|entry| entry == "assets")
                .count(),
            1,
        );
        assert!(fs.metadata(Path::new("/opt/assets/css/site.css")).is_ok());
    }

    #[tokio::test]
    async fn test_read_dir_rebases_entries_under_nested_mount_subdirectory() {
        let fs = MountFileSystem::new();

        let nested = mem_fs::FileSystem::default();
        nested.create_dir(Path::new("/css")).unwrap();
        nested
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/css/site.css"))
            .unwrap();

        fs.mount(Path::new("/opt/assets"), Arc::new(nested))
            .unwrap();

        let css_contents: Vec<PathBuf> = fs
            .read_dir(Path::new("/opt/assets/css"))
            .unwrap()
            .map(|entry| entry.unwrap().path)
            .collect();

        assert_eq!(
            css_contents,
            vec![PathBuf::from("/opt/assets/css/site.css")]
        );
    }

    #[tokio::test]
    async fn test_mount_with_source_path_exposes_subtree() {
        let fs = MountFileSystem::new();

        let source = mem_fs::FileSystem::default();
        source.create_dir(Path::new("/python")).unwrap();
        source
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/python/lib.py"))
            .unwrap();

        fs.mount_with_source(
            Path::new("/runtime"),
            Path::new("/python"),
            Arc::new(source),
        )
        .unwrap();

        assert!(fs.metadata(Path::new("/runtime/lib.py")).unwrap().is_file());
        assert_eq!(read_dir_names(&fs, "/runtime"), vec!["lib.py".to_string()]);
    }

    #[tokio::test]
    async fn test_nested_mount_inside_tree_preserves_sibling_files() {
        let fs = MountFileSystem::new();

        let python = mem_fs::FileSystem::default();
        create_dir_all(&python, Path::new("/usr/local/lib/python3.13/encodings"));
        python
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/usr/local/lib/python3.13/encodings/__init__.py"))
            .unwrap();

        let host = mem_fs::FileSystem::default();
        host.new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/marker.txt"))
            .unwrap();

        fs.mount(Path::new("/"), Arc::new(python)).unwrap();
        fs.mount(Path::new("/usr/local/lib/python3.13/test"), Arc::new(host))
            .unwrap();

        assert!(
            fs.metadata(Path::new("/usr/local/lib/python3.13/encodings/__init__.py"))
                .unwrap()
                .is_file()
        );
        assert!(
            fs.metadata(Path::new("/usr/local/lib/python3.13/test/marker.txt"))
                .unwrap()
                .is_file()
        );

        fs.new_open_options()
            .read(true)
            .open(Path::new("/usr/local/lib/python3.13/encodings/__init__.py"))
            .unwrap();
        fs.new_open_options()
            .read(true)
            .open(Path::new("/usr/local/lib/python3.13/test/marker.txt"))
            .unwrap();

        let mut entries = read_dir_names(&fs, "/usr/local/lib/python3.13");
        entries.sort();
        assert_eq!(entries, vec!["encodings".to_string(), "test".to_string()]);
    }

    #[tokio::test]
    async fn test_synthetic_parent_without_backing_dir_lists_child_mount() {
        let fs = MountFileSystem::new();
        fs.mount(Path::new("/"), Arc::new(mem_fs::FileSystem::default()))
            .unwrap();

        let child = mem_fs::FileSystem::default();
        child
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/marker.txt"))
            .unwrap();
        fs.mount(Path::new("/foo/bar"), Arc::new(child)).unwrap();

        let entries = read_dir_names(&fs, "/foo");
        assert_eq!(entries, vec!["bar".to_string()]);
    }

    #[tokio::test]
    async fn test_synthetic_parent_accepts_new_entries() {
        let fs = MountFileSystem::new();
        let root = Arc::new(mem_fs::FileSystem::default());
        fs.mount(Path::new("/"), root.clone()).unwrap();
        fs.mount(
            Path::new("/usr/local/ssl"),
            Arc::new(mem_fs::FileSystem::default()),
        )
        .unwrap();

        // `/usr` and `/usr/local` exist only because of the nested mount.
        fs.new_open_options()
            .write(true)
            .create(true)
            .open(Path::new("/usr/x"))
            .unwrap();
        assert_eq!(fs.create_dir(Path::new("/usr/local")), Ok(()));
        assert_eq!(fs.create_dir(Path::new("/usr/local/bin")), Ok(()));
        assert_eq!(fs.create_dir(Path::new("/usr/bin")), Ok(()));

        assert!(root.metadata(Path::new("/usr/x")).unwrap().is_file());
        assert!(root.metadata(Path::new("/usr/local/bin")).unwrap().is_dir());
        let mut entries = read_dir_names(&fs, "/usr");
        entries.sort();
        assert_eq!(entries, ["bin", "local", "x"]);
        let mut entries = read_dir_names(&fs, "/usr/local");
        entries.sort();
        assert_eq!(entries, ["bin", "ssl"]);
    }

    #[tokio::test]
    async fn test_import_mounts_allows_shared_prefix_without_exact_mount_conflict() {
        let primary = MountFileSystem::new();
        let bin = mem_fs::FileSystem::default();
        bin.new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/tool"))
            .unwrap();
        primary.mount(Path::new("/opt/bin"), Arc::new(bin)).unwrap();

        let injected = MountFileSystem::new();
        let assets = mem_fs::FileSystem::default();
        assets
            .new_open_options()
            .write(true)
            .create_new(true)
            .open(Path::new("/logo.svg"))
            .unwrap();
        injected
            .mount(Path::new("/opt/assets"), Arc::new(assets))
            .unwrap();

        primary
            .add_mount_entries_with_mode(
                injected.mount_entries(),
                super::ExactMountConflictMode::Fail,
            )
            .unwrap();

        assert!(primary.metadata(Path::new("/opt/bin/tool")).is_ok());
        assert!(primary.metadata(Path::new("/opt/assets/logo.svg")).is_ok());
    }

    #[tokio::test]
    async fn test_import_mounts_rejects_exact_mount_conflict() {
        let primary = MountFileSystem::new();
        primary
            .mount(
                Path::new("/opt/bin"),
                Arc::new(mem_fs::FileSystem::default()),
            )
            .unwrap();

        let injected = MountFileSystem::new();
        injected
            .mount(
                Path::new("/opt/bin"),
                Arc::new(mem_fs::FileSystem::default()),
            )
            .unwrap();

        assert_eq!(
            primary.add_mount_entries_with_mode(
                injected.mount_entries(),
                super::ExactMountConflictMode::Fail,
            ),
            Err(FsError::AlreadyExists)
        );
    }

    #[tokio::test]
    async fn test_new_filesystem() {
        let fs = gen_filesystem();
        assert!(
            fs.read_dir(Path::new("/test_new_filesystem")).is_ok(),
            "hostfs can read root"
        );
        let mut file_write = fs
            .new_open_options()
            .read(true)
            .write(true)
            .create_new(true)
            .open(Path::new("/test_new_filesystem/foo2.txt"))
            .unwrap();
        file_write.write_all(b"hello").await.unwrap();
        let _ = std::fs::remove_file("/test_new_filesystem/foo2.txt");
    }

    #[tokio::test]
    async fn test_create_dir() {
        let fs = gen_filesystem();

        assert_eq!(fs.create_dir(Path::new("/")), Ok(()));

        assert_eq!(fs.create_dir(Path::new("/test_create_dir")), Ok(()));

        assert_eq!(
            fs.create_dir(Path::new("/test_create_dir/foo")),
            Ok(()),
            "creating a directory",
        );

        let cur_dir = read_dir_names(&fs, "/test_create_dir");

        if !cur_dir.contains(&"foo".to_string()) {
            panic!("cur_dir does not contain foo: {cur_dir:#?}");
        }

        assert!(
            cur_dir.contains(&"foo".to_string()),
            "the root is updated and well-defined"
        );

        assert_eq!(
            fs.create_dir(Path::new("/test_create_dir/foo/bar")),
            Ok(()),
            "creating a sub-directory",
        );

        let foo_dir = read_dir_names(&fs, "/test_create_dir/foo");

        assert!(
            foo_dir.contains(&"bar".to_string()),
            "the foo directory is updated and well-defined"
        );

        let bar_dir = read_dir_names(&fs, "/test_create_dir/foo/bar");

        assert!(
            bar_dir.is_empty(),
            "the foo directory is updated and well-defined"
        );
        let _ = fs_extra::remove_items(&["/test_create_dir"]);
    }

    #[tokio::test]
    async fn test_remove_dir() {
        let fs = gen_filesystem();

        assert_eq!(
            fs.remove_dir(Path::new("/")),
            Err(FsError::PermissionDenied),
            "cannot remove the root directory",
        );

        assert_eq!(
            fs.remove_dir(Path::new("/foo")),
            Err(FsError::EntryNotFound),
            "cannot remove a directory that doesn't exist",
        );

        assert_eq!(fs.create_dir(Path::new("/test_remove_dir")), Ok(()));

        assert_eq!(
            fs.create_dir(Path::new("/test_remove_dir/foo")),
            Ok(()),
            "creating a directory",
        );

        assert_eq!(
            fs.create_dir(Path::new("/test_remove_dir/foo/bar")),
            Ok(()),
            "creating a sub-directory",
        );

        assert!(
            read_dir_names(&fs, "/test_remove_dir/foo").contains(&"bar".to_string()),
            "./foo/bar exists"
        );

        assert_eq!(
            fs.remove_dir(Path::new("/test_remove_dir/foo")),
            Err(FsError::DirectoryNotEmpty),
            "removing a directory that has children",
        );

        assert_eq!(
            fs.remove_dir(Path::new("/test_remove_dir/foo/bar")),
            Ok(()),
            "removing a sub-directory",
        );

        assert_eq!(
            fs.remove_dir(Path::new("/test_remove_dir/foo")),
            Ok(()),
            "removing a directory",
        );

        assert!(
            !read_dir_names(&fs, "/test_remove_dir").contains(&"foo".to_string()),
            "the foo directory still exists"
        );
    }

    fn read_dir_names(fs: &dyn crate::FileSystem, path: &str) -> Vec<String> {
        fs.read_dir(Path::new(path))
            .unwrap()
            .filter_map(|entry| Some(entry.ok()?.file_name().to_str()?.to_string()))
            .collect::<Vec<_>>()
    }

    fn create_dir_all(fs: &mem_fs::FileSystem, path: &Path) {
        let mut current = PathBuf::from("/");

        for component in path.iter().skip(1) {
            current.push(component);

            if fs.metadata(&current).is_err() {
                fs.create_dir(&current).unwrap();
            }
        }
    }

    #[tokio::test]
    async fn test_rename() {
        let fs = gen_filesystem();

        assert_eq!(
            fs.rename(Path::new("/"), Path::new("/bar")).await,
            Err(FsError::PermissionDenied),
            "renaming a directory that has no parent",
        );
        assert_eq!(
            fs.rename(Path::new("/foo"), Path::new("/")).await,
            Err(FsError::PermissionDenied),
            "renaming to the synthetic root directory is rejected",
        );

        assert_eq!(fs.create_dir(Path::new("/test_rename")), Ok(()));
        assert_eq!(fs.create_dir(Path::new("/test_rename/foo")), Ok(()));
        assert_eq!(fs.create_dir(Path::new("/test_rename/foo/qux")), Ok(()));

        assert_eq!(
            fs.rename(
                Path::new("/test_rename/foo"),
                Path::new("/test_rename/bar/baz")
            )
            .await,
            Err(FsError::EntryNotFound),
            "renaming to a directory that has parent that doesn't exist",
        );

        assert_eq!(fs.create_dir(Path::new("/test_rename/bar")), Ok(()));

        assert_eq!(
            fs.rename(Path::new("/test_rename/foo"), Path::new("/test_rename/bar"))
                .await,
            Ok(()),
            "renaming to a directory that has parent that exists",
        );

        assert!(
            fs.new_open_options()
                .write(true)
                .create_new(true)
                .open(Path::new("/test_rename/bar/hello1.txt"))
                .is_ok(),
            "creating a new file (`hello1.txt`)",
        );
        assert!(
            fs.new_open_options()
                .write(true)
                .create_new(true)
                .open(Path::new("/test_rename/bar/hello2.txt"))
                .is_ok(),
            "creating a new file (`hello2.txt`)",
        );

        let cur_dir = read_dir_names(&fs, "/test_rename");

        assert!(
            !cur_dir.contains(&"foo".to_string()),
            "the foo directory still exists"
        );

        assert!(
            cur_dir.contains(&"bar".to_string()),
            "the bar directory still exists"
        );

        let bar_dir = read_dir_names(&fs, "/test_rename/bar");

        if !bar_dir.contains(&"qux".to_string()) {
            println!("qux does not exist: {bar_dir:?}")
        }

        let qux_dir = read_dir_names(&fs, "/test_rename/bar/qux");

        assert!(qux_dir.is_empty(), "the qux directory is empty");

        assert!(
            read_dir_names(&fs, "/test_rename/bar").contains(&"hello1.txt".to_string()),
            "the /bar/hello1.txt file exists"
        );

        assert!(
            read_dir_names(&fs, "/test_rename/bar").contains(&"hello2.txt".to_string()),
            "the /bar/hello2.txt file exists"
        );

        assert_eq!(
            fs.create_dir(Path::new("/test_rename/foo")),
            Ok(()),
            "create ./foo again",
        );

        assert_eq!(
            fs.rename(
                Path::new("/test_rename/bar/hello2.txt"),
                Path::new("/test_rename/foo/world2.txt")
            )
            .await,
            Ok(()),
            "renaming (and moving) a file",
        );

        assert_eq!(
            fs.rename(
                Path::new("/test_rename/foo"),
                Path::new("/test_rename/bar/baz")
            )
            .await,
            Ok(()),
            "renaming a directory",
        );

        assert_eq!(
            fs.rename(
                Path::new("/test_rename/bar/hello1.txt"),
                Path::new("/test_rename/bar/world1.txt")
            )
            .await,
            Ok(()),
            "renaming a file (in the same directory)",
        );

        assert!(
            read_dir_names(&fs, "/test_rename").contains(&"bar".to_string()),
            "./bar exists"
        );

        assert!(
            read_dir_names(&fs, "/test_rename/bar").contains(&"baz".to_string()),
            "/bar/baz exists"
        );
        assert!(
            !read_dir_names(&fs, "/test_rename").contains(&"foo".to_string()),
            "foo does not exist anymore"
        );
        assert!(
            read_dir_names(&fs, "/test_rename/bar/baz").contains(&"world2.txt".to_string()),
            "/bar/baz/world2.txt exists"
        );
        assert!(
            read_dir_names(&fs, "/test_rename/bar").contains(&"world1.txt".to_string()),
            "/bar/world1.txt (ex hello1.txt) exists"
        );
        assert!(
            !read_dir_names(&fs, "/test_rename/bar").contains(&"hello1.txt".to_string()),
            "hello1.txt was moved"
        );
        assert!(
            !read_dir_names(&fs, "/test_rename/bar").contains(&"hello2.txt".to_string()),
            "hello2.txt was moved"
        );
        assert!(
            read_dir_names(&fs, "/test_rename/bar/baz").contains(&"world2.txt".to_string()),
            "world2.txt was moved to the correct place"
        );

        let _ = fs_extra::remove_items(&["/test_rename"]);
    }

    #[tokio::test]
    async fn cross_mount_file_rename_copies_and_removes_source() {
        let fs = MountFileSystem::new();
        let left = TmpFileSystem::new();
        let right = TmpFileSystem::new();

        left.new_open_options()
            .create(true)
            .write(true)
            .open(Path::new("/from.txt"))
            .unwrap();

        fs.mount(Path::new("/left"), Arc::new(left.clone()))
            .unwrap();
        fs.mount(Path::new("/right"), Arc::new(right.clone()))
            .unwrap();

        fs.rename(Path::new("/left/from.txt"), Path::new("/right/to.txt"))
            .await
            .unwrap();

        assert_eq!(
            left.metadata(Path::new("/from.txt")),
            Err(FsError::EntryNotFound)
        );
        assert!(right.metadata(Path::new("/to.txt")).unwrap().is_file());
    }

    #[tokio::test]
    async fn test_metadata() {
        use std::thread::sleep;
        use std::time::Duration;

        let fs = gen_filesystem();

        let root_metadata = fs.metadata(Path::new("/test_metadata")).unwrap();

        assert!(root_metadata.ft.dir);
        assert_eq!(root_metadata.accessed, root_metadata.created);
        assert_eq!(root_metadata.modified, root_metadata.created);
        assert!(root_metadata.modified > 0);

        assert_eq!(fs.create_dir(Path::new("/test_metadata/foo")), Ok(()));

        let foo_metadata = fs.metadata(Path::new("/test_metadata/foo"));
        assert!(foo_metadata.is_ok());
        let foo_metadata = foo_metadata.unwrap();

        assert!(foo_metadata.ft.dir);
        assert!(foo_metadata.accessed == foo_metadata.created);
        assert!(foo_metadata.modified == foo_metadata.created);
        assert!(foo_metadata.modified > 0);

        sleep(Duration::from_secs(3));

        assert_eq!(
            fs.rename(
                Path::new("/test_metadata/foo"),
                Path::new("/test_metadata/bar")
            )
            .await,
            Ok(())
        );

        let bar_metadata = fs.metadata(Path::new("/test_metadata/bar")).unwrap();
        assert!(bar_metadata.ft.dir);
        assert!(bar_metadata.accessed == foo_metadata.accessed);
        assert!(bar_metadata.created == foo_metadata.created);
        assert!(bar_metadata.modified > foo_metadata.modified);

        let root_metadata = fs.metadata(Path::new("/test_metadata/bar")).unwrap();
        assert!(
            root_metadata.modified > foo_metadata.modified,
            "the parent modified time was updated"
        );

        let _ = fs_extra::remove_items(&["/test_metadata"]);
    }

    #[tokio::test]
    async fn test_remove_file() {
        let fs = gen_filesystem();

        assert!(
            fs.new_open_options()
                .write(true)
                .create_new(true)
                .open(Path::new("/test_remove_file/foo.txt"))
                .is_ok(),
            "creating a new file",
        );

        assert!(read_dir_names(&fs, "/test_remove_file").contains(&"foo.txt".to_string()));

        assert_eq!(
            fs.remove_file(Path::new("/test_remove_file/foo.txt")),
            Ok(()),
            "removing a file that exists",
        );

        assert!(!read_dir_names(&fs, "/test_remove_file").contains(&"foo.txt".to_string()));

        assert_eq!(
            fs.remove_file(Path::new("/test_remove_file/foo.txt")),
            Err(FsError::EntryNotFound),
            "removing a file that doesn't exists",
        );

        let _ = fs_extra::remove_items(&["./test_remove_file"]);
    }

    #[tokio::test]
    async fn test_readdir() {
        let fs = gen_filesystem();

        assert_eq!(
            fs.create_dir(Path::new("/test_readdir/foo")),
            Ok(()),
            "creating `foo`"
        );
        assert_eq!(
            fs.create_dir(Path::new("/test_readdir/foo/sub")),
            Ok(()),
            "creating `sub`"
        );
        assert_eq!(
            fs.create_dir(Path::new("/test_readdir/bar")),
            Ok(()),
            "creating `bar`"
        );
        assert_eq!(
            fs.create_dir(Path::new("/test_readdir/baz")),
            Ok(()),
            "creating `bar`"
        );
        assert!(
            fs.new_open_options()
                .write(true)
                .create_new(true)
                .open(Path::new("/test_readdir/a.txt"))
                .is_ok(),
            "creating `a.txt`",
        );
        assert!(
            fs.new_open_options()
                .write(true)
                .create_new(true)
                .open(Path::new("/test_readdir/b.txt"))
                .is_ok(),
            "creating `b.txt`",
        );

        println!("fs: {fs:?}");

        let readdir = fs.read_dir(Path::new("/test_readdir"));

        assert!(readdir.is_ok(), "reading the directory `/test_readdir/`");

        let mut readdir = readdir.unwrap();

        let next = readdir.next().unwrap().unwrap();
        assert!(next.path.ends_with("foo"), "checking entry #1");
        println!("entry 1: {next:#?}");
        assert!(next.file_type().unwrap().is_dir(), "checking entry #1");

        let next = readdir.next().unwrap().unwrap();
        assert!(next.path.ends_with("bar"), "checking entry #2");
        assert!(next.file_type().unwrap().is_dir(), "checking entry #2");

        let next = readdir.next().unwrap().unwrap();
        assert!(next.path.ends_with("baz"), "checking entry #3");
        assert!(next.file_type().unwrap().is_dir(), "checking entry #3");

        let next = readdir.next().unwrap().unwrap();
        assert!(next.path.ends_with("a.txt"), "checking entry #2");
        assert!(next.file_type().unwrap().is_file(), "checking entry #4");

        let next = readdir.next().unwrap().unwrap();
        assert!(next.path.ends_with("b.txt"), "checking entry #2");
        assert!(next.file_type().unwrap().is_file(), "checking entry #5");

        if let Some(s) = readdir.next() {
            panic!("next: {s:?}");
        }

        let _ = fs_extra::remove_items(&["./test_readdir"]);
    }
}