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

use {
    crate::{
        builtin_environment::BuiltinEnvironment,
        klog,
        model::{
            binding::Binder,
            error::ModelError,
            hooks::HooksRegistration,
            model::{ComponentManagerConfig, Model, ModelParams},
            moniker::{AbsoluteMoniker, PartialMoniker, RelativeMoniker},
            resolver::ResolverRegistry,
            runner::Runner,
            testing::{echo_service::*, mocks::*, out_dir::OutDir, test_helpers::*},
        },
        startup,
    },
    cm_rust::*,
    fidl::endpoints::{self, create_proxy, ClientEnd, ServerEnd},
    fidl_fidl_examples_echo::{self as echo},
    fidl_fuchsia_component_runner as fcrunner,
    fidl_fuchsia_io::{
        DirectoryProxy, FileEvent, FileMarker, FileObject, FileProxy, NodeInfo, NodeMarker,
        CLONE_FLAG_SAME_RIGHTS, MODE_TYPE_DIRECTORY, MODE_TYPE_FILE, MODE_TYPE_SERVICE,
        OPEN_FLAG_CREATE, OPEN_FLAG_DESCRIBE, OPEN_RIGHT_READABLE, OPEN_RIGHT_WRITABLE,
    },
    fidl_fuchsia_sys2 as fsys, fuchsia_zircon as zx,
    fuchsia_zircon::HandleBased,
    futures::lock::Mutex,
    futures::prelude::*,
    std::{
        collections::{HashMap, HashSet},
        convert::{TryFrom, TryInto},
        default::Default,
        ffi::CString,
        path::{Path, PathBuf},
        ptr,
        sync::Arc,
    },
    tempfile::TempDir,
    vfs::directory::entry::DirectoryEntry,
};

/// Construct a capability path for the hippo service.
pub fn default_service_capability() -> CapabilityPath {
    "/svc/hippo".try_into().unwrap()
}

/// Construct a capability path for the hippo directory.
pub fn default_directory_capability() -> CapabilityPath {
    "/data/hippo".try_into().unwrap()
}

pub enum CheckUse {
    Protocol {
        path: CapabilityPath,
        should_succeed: bool,
    },
    Directory {
        path: CapabilityPath,
        file: PathBuf,
        should_succeed: bool,
    },
    Storage {
        path: CapabilityPath,
        type_: fsys::StorageType,
        // The relative moniker from the storage declaration to the use declaration. If None, this
        // storage use is expected to fail.
        storage_relation: Option<RelativeMoniker>,
        // The backing directory for this storage is in component manager's namsepace, not the
        // test's isolated test directory.
        from_cm_namespace: bool,
    },
    Event {
        names: Vec<CapabilityName>,
        should_be_allowed: bool,
    },
}

impl CheckUse {
    pub fn default_directory(should_succeed: bool) -> Self {
        Self::Directory {
            path: default_directory_capability(),
            file: PathBuf::from("hippo"),
            should_succeed,
        }
    }
}

/// Builder for setting up a new `RoutingTest` instance with a non-standard setup.
///
/// Use as follows:
///
/// ```
///   let universe = RoutingTestBuilder::new("root", components)
///       .add_hooks(...)
///       .add_outgoing_path(...)
///       .build();
/// ```
#[derive(Default)]
pub struct RoutingTestBuilder {
    root_component: String,
    components: Vec<(&'static str, ComponentDecl)>,
    additional_hooks: Vec<HooksRegistration>,
    outgoing_paths: HashMap<String, HashMap<CapabilityPath, Arc<dyn DirectoryEntry>>>,
    builtin_runners: HashMap<CapabilityName, Arc<dyn Runner>>,

    // Map of components that have a custom function serving the "outgoing" directory.
    // Other functions will receive a stock directory generated by `RoutingTest`.
    custom_outgoing_host_fns: HashMap<String, HostFn>,
}

impl RoutingTestBuilder {
    pub fn new(root_component: &str, components: Vec<(&'static str, ComponentDecl)>) -> Self {
        RoutingTestBuilder {
            root_component: root_component.to_string(),
            components,
            ..Default::default()
        }
    }

    pub fn add_hooks(mut self, mut hooks: Vec<HooksRegistration>) -> Self {
        self.additional_hooks.append(&mut hooks);
        self
    }

    /// Expose the given `DirectoryEntry` at the given path of the `component`'s outgoing
    /// directory.
    pub fn add_outgoing_path(
        mut self,
        component: &str,
        path: CapabilityPath,
        entry: Arc<dyn DirectoryEntry>,
    ) -> Self {
        self.outgoing_paths
            .entry(component.to_string())
            .or_insert_with(|| HashMap::new())
            .insert(path, entry);
        self
    }

    /// Add the given runner as a "builtin runner", offered to the root component
    /// under the given name.
    pub fn add_builtin_runner(
        mut self,
        name: impl Into<CapabilityName>,
        runner: Arc<dyn Runner>,
    ) -> Self {
        self.builtin_runners.insert(name.into(), runner);
        self
    }

    /// Request a custom outgoing directory host function.
    pub fn set_component_outgoing_host_fn(mut self, component: &str, function: HostFn) -> Self {
        self.custom_outgoing_host_fns.insert(component.to_string(), function);
        self
    }

    pub async fn build(self) -> RoutingTest {
        RoutingTest::from_builder(self).await
    }
}

/// A test for capability routing.
///
/// All string arguments are referring to component names, not URLs, ex: "a", not "test:///a" or
/// "test:///a_resolved".
pub struct RoutingTest {
    components: Vec<(&'static str, ComponentDecl)>,
    pub model: Arc<Model>,
    pub builtin_environment: BuiltinEnvironment,
    _echo_service: Arc<EchoService>,
    pub mock_runner: Arc<MockRunner>,
    _test_dir: TempDir,
    test_dir_proxy: DirectoryProxy,
    root_component_name: String,
}

impl RoutingTest {
    /// Initializes a new RoutingTest with a default setup.
    pub async fn new(root_component: &str, components: Vec<(&'static str, ComponentDecl)>) -> Self {
        RoutingTestBuilder::new(root_component, components).build().await
    }

    /// Construct a new `RoutingTest` from the given builder.
    async fn from_builder(mut builder: RoutingTestBuilder) -> Self {
        // Ensure that kernel logging has been set up
        klog::KernelLogger::init();

        let mut resolver = ResolverRegistry::new();
        let runner = Arc::new(MockRunner::new());

        let test_dir = TempDir::new_in("/tmp").expect("failed to create temp directory");

        // Create a directory for the components, starting with a single static file
        // "foo/hippo" in it.
        let test_dir_proxy = io_util::open_directory_in_namespace(
            test_dir.path().to_str().unwrap(),
            io_util::OPEN_RIGHT_READABLE | io_util::OPEN_RIGHT_WRITABLE,
        )
        .expect("failed to open temp directory");
        capability_util::create_static_file(&test_dir_proxy, Path::new("foo/hippo"), "hippo")
            .await
            .expect("could not create test file");

        // Create and populate an outgoing directory for each component.
        let mut mock_resolver = MockResolver::new();
        for (name, decl) in &builder.components {
            let host_fn = match builder.custom_outgoing_host_fns.remove(*name) {
                // If a custom outgoing HostFn was provided, use that.
                Some(host_fn) => host_fn,
                // Otherwise, create a default HostFn filled with sample data.
                None => Self::build_outgoing_dir(
                    decl,
                    &test_dir_proxy,
                    builder.outgoing_paths.remove(name as &str).unwrap_or_else(|| HashMap::new()),
                )
                .host_fn(),
            };
            runner.add_host_fn(&format!("test:///{}_resolved", name), host_fn);
            mock_resolver.add_component(name, decl.clone());
        }
        resolver.register("test".to_string(), Box::new(mock_resolver));

        // Set up runners for the system, including a default runner "test_runner"
        // backed by mock_runner.
        let mut builtin_runners = builder.builtin_runners;
        builtin_runners.insert(TEST_RUNNER_NAME.into(), runner.clone());

        let startup_args = startup::Arguments {
            use_builtin_process_launcher: false,
            root_component_url: "".to_string(),
            debug: false,
        };
        let echo_service = Arc::new(EchoService::new());

        let model = Arc::new(Model::new(ModelParams {
            root_component_url: format!("test:///{}", builder.root_component),
            root_resolver_registry: resolver,
        }));
        let builtin_environment = BuiltinEnvironment::new(
            &startup_args,
            &model,
            ComponentManagerConfig::default(),
            &builtin_runners,
        )
        .await
        .expect("builtin environment setup failed");

        model.root_realm.hooks.install(builder.additional_hooks).await;
        model.root_realm.hooks.install(echo_service.hooks()).await;

        Self {
            components: builder.components,
            model,
            builtin_environment,
            _echo_service: echo_service,
            mock_runner: runner,
            _test_dir: test_dir,
            test_dir_proxy,
            root_component_name: builder.root_component,
        }
    }

    /// Installs a new directory at /hippo in our namespace. Does nothing if this directory already
    /// exists.
    pub fn install_hippo_dir(&self, path: &str) {
        let (client_chan, server_chan) = zx::Channel::create().unwrap();

        let mut ns_ptr: *mut fdio::fdio_sys::fdio_ns_t = ptr::null_mut();
        let status = unsafe { fdio::fdio_sys::fdio_ns_get_installed(&mut ns_ptr) };
        if status != zx::sys::ZX_OK {
            panic!(
                "bad status returned for fdio_ns_get_installed: {}",
                zx::Status::from_raw(status)
            );
        }
        let cstr = CString::new(path).unwrap();
        let status =
            unsafe { fdio::fdio_sys::fdio_ns_bind(ns_ptr, cstr.as_ptr(), client_chan.into_raw()) };
        if status != zx::sys::ZX_OK && status != zx::sys::ZX_ERR_ALREADY_EXISTS {
            panic!("bad status returned for fdio_ns_bind: {}", zx::Status::from_raw(status));
        }
        let mut out_dir = OutDir::new();
        Self::install_default_out_files(&mut out_dir);
        out_dir.add_directory_proxy(&self.test_dir_proxy);
        out_dir.host_fn()(ServerEnd::new(server_chan));
    }

    /// Creates a static file at the given path in the temp directory.
    pub async fn create_static_file(
        &self,
        path: &Path,
        contents: &str,
    ) -> Result<(), anyhow::Error> {
        capability_util::create_static_file(&self.test_dir_proxy, path, contents).await
    }

    /// Set up the given OutDir, installing a set of files assumed to exist by
    /// many tests:
    ///   - A file `/svc/foo` implementing `fidl.examples.echo.Echo`.
    ///   - A static file `/svc/file`, containing the string "hippos" encoded as UTF-8.
    fn install_default_out_files(dir: &mut OutDir) {
        // Add "/svc/foo", providing an echo server.
        dir.add_echo_service(CapabilityPath::try_from("/svc/foo").unwrap());

        // Add "/svc/file", providing a read-only file.
        dir.add_static_file(CapabilityPath::try_from("/svc/file").unwrap(), "hippos");
    }

    /// Creates a dynamic child `child_decl` in `moniker`'s `collection`.
    pub async fn create_dynamic_child<'a>(
        &'a self,
        moniker: AbsoluteMoniker,
        collection: &'a str,
        decl: ChildDecl,
    ) {
        let component_name = self.bind_instance(&moniker).await.expect("bind instance failed");
        let component_resolved_url = Self::resolved_url(&component_name);
        Self::check_namespace(component_name, &self.mock_runner, self.components.clone()).await;
        let namespace = self
            .mock_runner
            .get_namespace(&component_resolved_url)
            .expect("could not find child namespace");
        capability_util::call_create_child(&namespace, collection, decl).await;
    }

    /// Deletes a dynamic child `child_decl` in `moniker`'s `collection`, waiting for destruction
    /// to complete.
    pub async fn destroy_dynamic_child<'a>(
        &'a self,
        moniker: AbsoluteMoniker,
        collection: &'a str,
        name: &'a str,
    ) {
        let realm = self.model.look_up_realm(&moniker).await.expect("failed to look up realm");
        self.model.bind(&realm.abs_moniker).await.expect("bind instance failed");
        let partial_moniker = PartialMoniker::new(name.to_string(), Some(collection.to_string()));
        let nf =
            realm.remove_dynamic_child(&partial_moniker).await.expect("failed to remove child");
        // Wait for destruction to fully complete.
        nf.await.expect("failed to destroy child");
    }

    /// Checks a `use` declaration at `moniker` by trying to use `capability`.
    pub async fn check_use(&self, moniker: AbsoluteMoniker, check: CheckUse) {
        let component_name = self.bind_instance(&moniker).await.expect("bind instance failed");
        let component_resolved_url = Self::resolved_url(&component_name);
        let namespace = self
            .mock_runner
            .get_namespace(&component_resolved_url)
            .expect("could not find child namespace");
        Self::check_namespace(component_name, &self.mock_runner, self.components.clone()).await;
        match check {
            CheckUse::Protocol { path, should_succeed } => {
                capability_util::call_echo_svc_from_namespace(&namespace, path, should_succeed)
                    .await;
            }
            CheckUse::Directory { path, file, should_succeed } => {
                capability_util::read_data_from_namespace(&namespace, path, &file, should_succeed)
                    .await
            }
            CheckUse::Storage { type_: fsys::StorageType::Meta, storage_relation, .. } => {
                capability_util::write_file_to_meta_storage(
                    &self.model,
                    moniker,
                    storage_relation.is_some(),
                )
                .await;
                if let Some(relative_moniker) = storage_relation {
                    capability_util::check_file_in_storage(
                        fsys::StorageType::Meta,
                        relative_moniker,
                        &self.test_dir_proxy,
                    )
                    .await;
                }
            }
            CheckUse::Storage { path, type_, storage_relation, from_cm_namespace } => {
                capability_util::write_file_to_storage(
                    &namespace,
                    path,
                    storage_relation.is_some(),
                )
                .await;

                if let Some(relative_moniker) = storage_relation {
                    if from_cm_namespace {
                        // Check for the file in the /tmp in the test's namespace
                        let tmp_proxy = io_util::open_directory_in_namespace(
                            "/tmp",
                            io_util::OPEN_RIGHT_READABLE,
                        )
                        .expect("failed to open /tmp");
                        capability_util::check_file_in_storage(type_, relative_moniker, &tmp_proxy)
                            .await;
                    } else {
                        // Check for the file in the test's isolated test directory
                        capability_util::check_file_in_storage(
                            type_,
                            relative_moniker,
                            &self.test_dir_proxy,
                        )
                        .await;
                    }
                }
            }
            CheckUse::Event { names, should_be_allowed } => {
                // Fails if the component did not use the protocol EventSource or if the event is
                // not allowed.
                capability_util::subscribe_to_event_stream(&namespace, should_be_allowed, names)
                    .await;
            }
        }
    }

    /// Checks using a capability from a component's exposed directory.
    pub async fn check_use_exposed_dir(&self, moniker: AbsoluteMoniker, check: CheckUse) {
        match check {
            CheckUse::Protocol { path, should_succeed } => {
                capability_util::call_echo_svc_from_exposed_dir(
                    path,
                    &moniker,
                    &self.model,
                    should_succeed,
                )
                .await;
            }
            CheckUse::Directory { path, file, should_succeed } => {
                capability_util::read_data_from_exposed_dir(
                    path,
                    &file,
                    &moniker,
                    &self.model,
                    should_succeed,
                )
                .await;
            }
            CheckUse::Storage { .. } => {
                panic!("storage capabilities can't be exposed");
            }
            CheckUse::Event { .. } => {
                panic!("event capabilities can't be exposed");
            }
        }
    }

    /// Lists the contents of a storage directory.
    pub async fn list_directory_in_storage(
        &self,
        relation: RelativeMoniker,
        relative_path: &str,
    ) -> Vec<String> {
        let mut dir_path = capability_util::generate_storage_path(None, &relation);
        if !relative_path.is_empty() {
            dir_path.push(relative_path);
        }
        if !dir_path.parent().is_none() {
            let dir_proxy = io_util::open_directory(
                &self.test_dir_proxy,
                &dir_path,
                io_util::OPEN_RIGHT_READABLE,
            )
            .expect("failed to open directory");
            list_directory(&dir_proxy).await
        } else {
            list_directory(&self.test_dir_proxy).await
        }
    }

    /// check_namespace will ensure that the paths in `namespaces` for `component_name` match the use
    /// declarations for the the component by the same name in `components`.
    async fn check_namespace(
        component_name: String,
        runner: &MockRunner,
        components: Vec<(&str, ComponentDecl)>,
    ) {
        let (_, decl) = components
            .into_iter()
            .find(|(name, _)| name == &component_name)
            .expect("component not in component decl list");
        // Two services installed into the same dir will cause duplicates, so use a HashSet to remove
        // them.
        let expected_paths_hs: HashSet<String> = decl
            .uses
            .into_iter()
            .filter_map(|u| match u {
                UseDecl::Directory(d) => Some(d.target_path.to_string()),
                UseDecl::Service(s) => Some(s.target_path.to_string()),
                UseDecl::Protocol(s) => Some(s.target_path.dirname),
                UseDecl::Storage(UseStorageDecl::Data(p)) => Some(p.to_string()),
                UseDecl::Storage(UseStorageDecl::Cache(p)) => Some(p.to_string()),
                UseDecl::Storage(UseStorageDecl::Meta) => None,
                UseDecl::Runner(_) | UseDecl::Event(_) => None,
            })
            .collect();
        let mut expected_paths = vec![];
        expected_paths.extend(expected_paths_hs.into_iter());

        // Get the paths in the component's namespace.
        let mut actual_paths: Vec<String> = runner
            .get_namespace(&format!("test:///{}_resolved", component_name))
            .expect("component not in namespace")
            .lock()
            .await
            .iter()
            .map(|entry| entry.path.as_ref().unwrap().clone())
            .collect();

        expected_paths.sort_unstable();
        actual_paths.sort_unstable();
        assert_eq!(expected_paths, actual_paths);
    }

    /// Checks a `use /svc/fuchsia.sys2.Realm` declaration at `moniker` by calling
    /// `BindChild`.
    pub async fn check_use_realm(
        &self,
        moniker: AbsoluteMoniker,
        bind_calls: Arc<Mutex<Vec<String>>>,
    ) {
        let component_name = self.bind_instance(&moniker).await.expect("bind instance failed");
        let component_resolved_url = Self::resolved_url(&component_name);
        let path = "/svc/fuchsia.sys2.Realm".try_into().unwrap();
        Self::check_namespace(component_name, &self.mock_runner, self.components.clone()).await;
        capability_util::call_realm_svc(
            path,
            &component_resolved_url,
            &self.mock_runner.get_namespace(&component_resolved_url).unwrap(),
            bind_calls.clone(),
        )
        .await;
    }

    /// Checks that a use declaration of `path` at `moniker` can be opened with
    /// Fuchsia file operations.
    pub async fn check_open_file(&self, moniker: AbsoluteMoniker, path: CapabilityPath) {
        let component_name = self.bind_instance(&moniker).await.expect("bind instance failed");
        let component_resolved_url = Self::resolved_url(&component_name);
        Self::check_namespace(component_name, &self.mock_runner, self.components.clone()).await;
        let namespace = self.mock_runner.get_namespace(&component_resolved_url).unwrap();
        capability_util::call_file_svc_from_namespace(&namespace, path).await;
    }

    /// Build an outgoing directory for the given component.
    fn build_outgoing_dir(
        decl: &ComponentDecl,
        test_dir_proxy: &DirectoryProxy,
        mut outgoing_paths: HashMap<CapabilityPath, Arc<dyn DirectoryEntry>>,
    ) -> OutDir {
        // if this decl is offering/exposing something from `Self`, let's host it
        let mut out_dir = OutDir::new();
        for expose in decl.exposes.iter() {
            match expose {
                ExposeDecl::Service(_) => panic!("service capability unsupported"),
                ExposeDecl::Protocol(s) if s.source == ExposeSource::Self_ => {
                    Self::install_default_out_files(&mut out_dir);
                }
                ExposeDecl::Directory(d) if d.source == ExposeSource::Self_ => {
                    out_dir.add_directory_proxy(test_dir_proxy)
                }
                _ => (),
            }
        }
        for offer in decl.offers.iter() {
            match offer {
                OfferDecl::Service(_) => panic!("service capability unsupported"),
                OfferDecl::Protocol(s) if s.source == OfferServiceSource::Self_ => {
                    Self::install_default_out_files(&mut out_dir);
                }
                OfferDecl::Directory(d) if d.source == OfferDirectorySource::Self_ => {
                    out_dir.add_directory_proxy(test_dir_proxy)
                }
                _ => (),
            }
        }
        for storage in decl.storage.iter() {
            // Storage capabilities can have a source of "self", so there are situations we want to
            // test where a storage capability is offered and used and there's no directory
            // capability in the manifest, so we must host the directory structure for this case in
            // addition to directory offers.
            if storage.source == StorageDirectorySource::Self_ {
                out_dir.add_directory_proxy(test_dir_proxy)
            }
        }
        // Add any user-specific DirectoryEntry objects into the outgoing namespace.
        for (path, entry) in outgoing_paths.drain() {
            out_dir.add_entry(path, entry);
        }

        out_dir
    }

    /// Atempt to bind the instance associated with the given moniker.
    ///
    /// On success, returns the short name of the component.
    pub async fn bind_instance(&self, moniker: &AbsoluteMoniker) -> Result<String, anyhow::Error> {
        self.model.bind(moniker).await?;
        Ok(match moniker.path().last() {
            Some(part) => part.name().to_string(),
            None => self.root_component_name.to_string(),
        })
    }

    /// Wait for the given component to start running.
    ///
    /// We define "running" as "the components outgoing directory has responded to a simple
    /// request", which the MockRunner supports.
    pub async fn wait_for_component_start(&self, component: &AbsoluteMoniker) {
        // Lookup, bind, and open a connection to the realm's outgoing directory.
        let (dir_proxy, server_end) =
            fidl::endpoints::create_proxy::<fidl_fuchsia_io::DirectoryMarker>().unwrap();
        let realm = self.model.look_up_realm(component).await.expect("lookup root realm failed");
        self.model
            .bind(&realm.abs_moniker)
            .await
            .expect("failed to bind to realm")
            .open_outgoing(
                fidl_fuchsia_io::OPEN_RIGHT_READABLE,
                fidl_fuchsia_io::MODE_TYPE_DIRECTORY,
                PathBuf::from("/."),
                server_end.into_channel(),
            )
            .await
            .expect("failed to open realm's outgoing directory");

        // Ensure we can successfully talk to the directory.
        dir_proxy.sync().await.expect("could not communicate with directory");
    }

    pub fn resolved_url(component_name: &str) -> String {
        format!("test:///{}_resolved", component_name)
    }
}

/// Contains functions to use capabilities in routing tests.
pub mod capability_util {
    use {
        super::*, crate::model::events::source_factory::EVENT_SOURCE_SYNC_SERVICE_PATH,
        cm_rust::NativeIntoFidl, fidl::endpoints::ServiceMarker,
        fidl_fuchsia_sys2::BlockingEventSourceMarker, std::path::PathBuf,
    };

    /// Looks up `resolved_url` in the namespace, and attempts to read ${path}/hippo. The file
    /// should contain the string "hippo".
    pub async fn read_data_from_namespace(
        namespace: &ManagedNamespace,
        path: CapabilityPath,
        file: &Path,
        should_succeed: bool,
    ) {
        let path = path.to_string();
        let dir_proxy = get_dir_from_namespace(namespace, &path).await;
        let file_proxy =
            io_util::open_file(&dir_proxy, file, OPEN_RIGHT_READABLE).expect("failed to open file");
        let res = io_util::read_file(&file_proxy).await;

        match should_succeed {
            true => assert_eq!("hippo", res.expect("failed to read file")),
            false => assert!(res.is_err(), "read file successfully when it should fail"),
        }
    }

    pub async fn write_file_to_storage(
        namespace: &ManagedNamespace,
        path: CapabilityPath,
        should_succeed: bool,
    ) {
        let dir_string = path.to_string();
        let dir_proxy = get_dir_from_namespace(namespace, dir_string.as_str()).await;
        write_hippo_file_to_directory(&dir_proxy, should_succeed).await;
    }

    pub async fn write_file_to_meta_storage(
        model: &Arc<Model>,
        moniker: AbsoluteMoniker,
        should_succeed: bool,
    ) {
        let realm = model.look_up_realm(&moniker).await.expect("failed to look up realm");
        let meta_dir_res = realm.resolve_meta_dir().await;
        match (meta_dir_res, should_succeed) {
            (Ok(Some(meta_dir)), true) => write_hippo_file_to_directory(&meta_dir, true).await,
            (Err(ModelError::CapabilityDiscoveryError { .. }), false) => (),
            (Err(ModelError::StorageError { .. }), false) => (),
            (Ok(Some(_)), false) => panic!("meta dir present when usage was expected to fail"),
            (Ok(None), true) => panic!("meta dir missing when usage was expected to succeed"),
            (Ok(None), false) => panic!("meta dir missing when resolution should return an error"),
            (Err(_), true) => panic!("meta dir resolution failed usage was expected to succeed"),
            (Err(_), false) => panic!("unexpected error when attempting meta dir resolution"),
        }
    }

    async fn write_hippo_file_to_directory(dir_proxy: &DirectoryProxy, should_succeed: bool) {
        let (file_proxy, server_end) = create_proxy::<FileMarker>().unwrap();
        let flags = OPEN_RIGHT_WRITABLE | OPEN_FLAG_CREATE;
        dir_proxy
            .open(flags, MODE_TYPE_FILE, "hippos", ServerEnd::new(server_end.into_channel()))
            .expect("failed to open file on storage");
        let res = file_proxy.write(b"hippos can be stored here").await;
        match (res, should_succeed) {
            (Ok((s, _)),true) => assert_eq!(zx::Status::OK, zx::Status::from_raw(s)),
            (Err(_),false) => (),
            (Ok((s, _)),false) => panic!("we shouldn't be able to access storage, but we opened a file! failed write with status {}", zx::Status::from_raw(s)),
            (Err(e),true) => panic!("failed to write to file when we expected to be able to! {:?}", e),
        }
    }

    /// Create a file with the given contents in the test dir, along with any subdirectories
    /// required.
    pub(super) async fn create_static_file(
        root: &DirectoryProxy,
        path: &Path,
        contents: &str,
    ) -> Result<(), anyhow::Error> {
        // Open file, and create subdirectories if required.
        let file_proxy = if let Some(directory) = path.parent() {
            let subdir = io_util::create_sub_directories(root, directory)
                .map_err(|e| e.context(format!("failed to create subdirs for {:?}", path)))?;
            io_util::open_file(
                &subdir,
                &PathBuf::from(path.file_name().unwrap()),
                OPEN_RIGHT_WRITABLE | OPEN_FLAG_CREATE,
            )?
        } else {
            io_util::open_file(root, path, OPEN_RIGHT_WRITABLE | OPEN_FLAG_CREATE)?
        };

        // Write contents.
        io_util::write_file(&file_proxy, contents).await
    }

    pub async fn check_file_in_storage(
        type_: fsys::StorageType,
        relation: RelativeMoniker,
        test_dir_proxy: &DirectoryProxy,
    ) {
        let mut dir_path = generate_storage_path(Some(type_), &relation);
        dir_path.push("hippos");
        let file_proxy =
            io_util::open_file(&test_dir_proxy, &dir_path, io_util::OPEN_RIGHT_READABLE)
                .expect("failed to open file");
        let res = io_util::read_file(&file_proxy).await;
        assert_eq!("hippos can be stored here".to_string(), res.expect("failed to read file"));
    }

    async fn connect_to_svc_in_namespace<T: ServiceMarker>(
        namespace: &ManagedNamespace,
        path: &CapabilityPath,
    ) -> T::Proxy {
        let dir_proxy = get_dir_from_namespace(namespace, &path.dirname).await;
        let node_proxy = io_util::open_node(
            &dir_proxy,
            &Path::new(&path.basename),
            OPEN_RIGHT_READABLE,
            MODE_TYPE_SERVICE,
        )
        .expect("failed to open echo service");
        let client_end = ClientEnd::<T>::new(node_proxy.into_channel().unwrap().into_zx_channel());
        client_end.into_proxy().unwrap()
    }

    /// Verifies that it's possible to subscribe to the given `events` by connecting to an
    /// `BlockingEventSource` on the given `namespace`. Used to test eventcapability routing.
    /// Testing of usage of the stream lives in the integration tests in:
    /// //src/sys/component_manager/tests/events/integration_test.rs
    pub async fn subscribe_to_event_stream(
        namespace: &ManagedNamespace,
        should_succeed: bool,
        events: Vec<CapabilityName>,
    ) {
        let event_source_proxy = connect_to_svc_in_namespace::<BlockingEventSourceMarker>(
            namespace,
            &*EVENT_SOURCE_SYNC_SERVICE_PATH,
        )
        .await;
        let event_names: Vec<String> = events.into_iter().map(|event| event.to_string()).collect();
        let (client_end, _stream) =
            fidl::endpoints::create_request_stream::<fsys::EventStreamMarker>()
                .expect("create client");
        let res = event_source_proxy
            .subscribe(&mut event_names.iter().map(|e| e.as_ref()), client_end)
            .await
            .expect("failed to use service");
        event_source_proxy.start_component_tree().await.expect("failed to start component tree");
        assert_eq!(res.is_ok(), should_succeed);
    }

    /// Looks up `resolved_url` in the namespace, and attempts to use `path`. Expects the service
    /// to be fidl.examples.echo.Echo.
    pub async fn call_echo_svc_from_namespace(
        namespace: &ManagedNamespace,
        path: CapabilityPath,
        should_succeed: bool,
    ) {
        let echo_proxy = connect_to_svc_in_namespace::<echo::EchoMarker>(namespace, &path).await;
        let res = echo_proxy.echo_string(Some("hippos")).await;

        match should_succeed {
            true => {
                assert_eq!(res.expect("failed to use echo service"), Some("hippos".to_string()))
            }
            false => {
                let err = res.expect_err("used echo service successfully when it should fail");
                assert!(err.is_closed(), "expected file closed error, got: {:?}", err);
            }
        }
    }

    /// Looks up `resolved_url` in the namespace, and attempts to use `path`.
    /// Expects the service to work like a fuchsia.io service, and respond with
    /// an OnOpen event when opened with OPEN_FLAG_DESCRIBE.
    pub async fn call_file_svc_from_namespace(namespace: &ManagedNamespace, path: CapabilityPath) {
        let dir_proxy = get_dir_from_namespace(namespace, &path.dirname).await;
        let node_proxy = io_util::open_node(
            &dir_proxy,
            &Path::new(&path.basename),
            OPEN_RIGHT_READABLE | OPEN_FLAG_DESCRIBE,
            // This should be MODE_TYPE_SERVICE, but we implement the underlying
            // service as a file for convenience in testing.
            MODE_TYPE_FILE,
        )
        .expect("failed to open file service");

        let file_proxy = FileProxy::new(node_proxy.into_channel().unwrap());
        let mut event_stream = file_proxy.take_event_stream();
        let event = event_stream.try_next().await.unwrap();
        let FileEvent::OnOpen_ { s, info } = event.expect("failed to received file event");
        assert_eq!(s, zx::sys::ZX_OK);
        assert_eq!(
            *info.expect("failed to receive node info"),
            NodeInfo::File(FileObject { event: None, stream: None })
        );
    }

    /// Attempts to read ${path}/hippo in `abs_moniker`'s exposed directory. The file should
    /// contain the string "hippo".
    pub async fn read_data_from_exposed_dir<'a>(
        path: CapabilityPath,
        file: &Path,
        abs_moniker: &'a AbsoluteMoniker,
        model: &'a Arc<Model>,
        should_succeed: bool,
    ) {
        let (node_proxy, server_end) = endpoints::create_proxy::<NodeMarker>().unwrap();
        open_exposed_dir(&path, abs_moniker, model, MODE_TYPE_DIRECTORY, server_end).await;
        let dir_proxy = DirectoryProxy::new(node_proxy.into_channel().unwrap());
        let file_proxy = io_util::open_file(&dir_proxy, &file, OPEN_RIGHT_READABLE);

        match file_proxy {
            Ok(fp) => {
                let res = io_util::read_file(&fp).await;
                match should_succeed {
                    true => assert_eq!("hippo", res.expect("failed to read file")),
                    false => assert!(res.is_err(), "read file successfully when it should fail"),
                }
            }
            Err(_) => {
                assert!(!should_succeed, file_proxy.expect("failed to open file when it should"))
            }
        }
    }

    /// Attempts to use the fidl.examples.echo.Echo service at `path` in `abs_moniker`'s exposed
    /// directory.
    pub async fn call_echo_svc_from_exposed_dir<'a>(
        path: CapabilityPath,
        abs_moniker: &'a AbsoluteMoniker,
        model: &'a Arc<Model>,
        should_succeed: bool,
    ) {
        let (node_proxy, server_end) = endpoints::create_proxy::<NodeMarker>().unwrap();
        open_exposed_dir(&path, abs_moniker, model, MODE_TYPE_SERVICE, server_end).await;
        let echo_proxy = echo::EchoProxy::new(node_proxy.into_channel().unwrap());
        let res = echo_proxy.echo_string(Some("hippos")).await;
        match should_succeed {
            true => {
                assert_eq!(res.expect("failed to use echo service"), Some("hippos".to_string()))
            }
            false => {
                let err = res.expect_err("used echo service successfully when it should fail");
                assert!(err.is_closed(), "expected file closed error, got: {:?}", err);
            }
        }
    }

    /// Looks up `resolved_url` in the namespace, and attempts to use `path`. Expects the service
    /// to be fuchsia.sys2.Realm.
    pub async fn call_realm_svc(
        path: CapabilityPath,
        resolved_url: &str,
        namespace: &ManagedNamespace,
        bind_calls: Arc<Mutex<Vec<String>>>,
    ) {
        let dir_proxy = get_dir_from_namespace(namespace, &path.dirname).await;
        let node_proxy = io_util::open_node(
            &dir_proxy,
            &Path::new(&path.basename),
            OPEN_RIGHT_READABLE,
            MODE_TYPE_SERVICE,
        )
        .expect("failed to open realm service");
        let realm_proxy = fsys::RealmProxy::new(node_proxy.into_channel().unwrap());
        let mut child_ref = fsys::ChildRef { name: "my_child".to_string(), collection: None };
        let (_client_chan, server_chan) = zx::Channel::create().unwrap();
        let exposed_capabilities = ServerEnd::new(server_chan);
        let res = realm_proxy.bind_child(&mut child_ref, exposed_capabilities).await;

        // Check for side effects: realm service should have received the `bind_child` call.
        let _ = res.expect("failed to use realm service");
        let bind_url =
            format!("test:///{}_resolved", bind_calls.lock().await.last().expect("no bind call"));
        assert_eq!(bind_url, resolved_url);
    }

    /// Call `fuchsia.sys2.Realm.CreateChild` to create a dynamic child.
    pub async fn call_create_child<'a>(
        namespace: &ManagedNamespace,
        collection: &'a str,
        child_decl: ChildDecl,
    ) {
        let path: CapabilityPath = "/svc/fuchsia.sys2.Realm".try_into().expect("no realm service");
        let dir_proxy = get_dir_from_namespace(namespace, &path.dirname).await;
        let node_proxy = io_util::open_node(
            &dir_proxy,
            &Path::new(&path.basename),
            OPEN_RIGHT_READABLE,
            MODE_TYPE_SERVICE,
        )
        .expect("failed to open realm service");
        let realm_proxy = fsys::RealmProxy::new(node_proxy.into_channel().unwrap());
        let mut collection_ref = fsys::CollectionRef { name: collection.to_string() };
        let child_decl = child_decl.native_into_fidl();
        let res = realm_proxy.create_child(&mut collection_ref, child_decl).await;
        let _ = res.expect("failed to create child");
    }

    /// Call `fuchsia.sys2.Realm.DestroyChild` to destroy a dynamic child, waiting for
    /// destruction to complete.
    pub async fn call_destroy_child<'a>(
        namespace: &ManagedNamespace,
        collection: &'a str,
        name: &'a str,
    ) {
        let path: CapabilityPath = "/svc/fuchsia.sys2.Realm".try_into().expect("no realm service");
        let dir_proxy = get_dir_from_namespace(namespace, &path.dirname).await;
        let node_proxy = io_util::open_node(
            &dir_proxy,
            &Path::new(&path.basename),
            OPEN_RIGHT_READABLE,
            MODE_TYPE_SERVICE,
        )
        .expect("failed to open realm service");
        let realm_proxy = fsys::RealmProxy::new(node_proxy.into_channel().unwrap());
        let mut child_ref =
            fsys::ChildRef { collection: Some(collection.to_string()), name: name.to_string() };
        let res = realm_proxy.destroy_child(&mut child_ref).await;
        let _ = res.expect("failed to destroy child");
    }

    /// Returns a cloned DirectoryProxy to the dir `dir_string` inside the namespace of
    /// `resolved_url`.
    pub async fn get_dir_from_namespace(
        namespace: &ManagedNamespace,
        dir_string: &str,
    ) -> DirectoryProxy {
        let mut ns = namespace.lock().await;

        // Find the index of our directory in the namespace, and remove the directory and path. The
        // path is removed so that the paths/dirs aren't shuffled in the namespace.
        let index = ns
            .iter()
            .position(|entry| entry.path.as_ref().unwrap() == dir_string)
            .expect(&format!("didn't find dir {}", dir_string));
        let entry = ns.remove(index);

        // Clone our directory, and then put the directory and path back on the end of the namespace so
        // that the namespace is (somewhat) unmodified.
        let dir_proxy = entry.directory.unwrap().into_proxy().unwrap();
        let dir_proxy_clone = io_util::clone_directory(&dir_proxy, CLONE_FLAG_SAME_RIGHTS).unwrap();
        ns.push(fcrunner::ComponentNamespaceEntry {
            path: entry.path,
            directory: Some(ClientEnd::new(dir_proxy.into_channel().unwrap().into_zx_channel())),
        });

        dir_proxy_clone
    }

    /// Open the exposed dir for `abs_moniker`.
    async fn open_exposed_dir<'a>(
        path: &'a CapabilityPath,
        abs_moniker: &'a AbsoluteMoniker,
        model: &'a Arc<Model>,
        open_mode: u32,
        server_end: ServerEnd<NodeMarker>,
    ) {
        let realm = model
            .look_up_realm(abs_moniker)
            .await
            .expect(&format!("realm not found {}", abs_moniker));
        model.bind(abs_moniker).await.expect("failed to bind instance");
        let execution = realm.lock_execution().await;
        let runtime = execution.runtime.as_ref().expect("not resolved");
        let flags = OPEN_RIGHT_READABLE;

        let vns_path = to_fvfs_path(path);
        runtime.exposed_dir.open(flags, open_mode, vns_path, server_end);
    }

    // This function should reproduce the logic of `crate::storage::generate_storage_path`
    pub fn generate_storage_path(
        type_: Option<fsys::StorageType>,
        relative_moniker: &RelativeMoniker,
    ) -> PathBuf {
        assert!(relative_moniker.up_path().is_empty());
        let mut down_path = relative_moniker.down_path().iter();
        let mut dir_path = vec![];
        if let Some(p) = down_path.next() {
            dir_path.push(p.as_str().to_string());
        }
        while let Some(p) = down_path.next() {
            dir_path.push("children".to_string());
            dir_path.push(p.as_str().to_string());
        }
        match type_ {
            Some(fsys::StorageType::Data) => dir_path.push("data".to_string()),
            Some(fsys::StorageType::Cache) => dir_path.push("cache".to_string()),
            Some(fsys::StorageType::Meta) => dir_path.push("meta".to_string()),
            None => {}
        }
        dir_path.into_iter().collect()
    }

    /// Function to convert a CapabilityPath to a pseudo_fs_mt::Path
    fn to_fvfs_path(path: &CapabilityPath) -> vfs::path::Path {
        let full_path = format!("{}/{}", path.dirname, path.basename);
        let split_string = full_path.split('/').filter(|s| !s.is_empty()).collect::<Vec<_>>();
        vfs::path::Path::validate_and_split(split_string.join("/"))
            .expect("Failed to validate and split path")
    }
}