// 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::runner::BuiltinRunnerFactory,
        builtin_environment::{BuiltinEnvironment, BuiltinEnvironmentBuilder},
        model::{
            component::{ComponentInstance, InstanceState, StartReason},
            error::ModelError,
            hooks::HooksRegistration,
            model::Model,
            starter::Starter,
            testing::{echo_service::*, mocks::*, out_dir::OutDir, test_helpers::*},
        },
    },
    ::routing::{
        component_id_index::ComponentInstanceId,
        component_instance::ComponentInstanceInterface,
        config::{
            AllowlistEntry, CapabilityAllowlistKey, ChildPolicyAllowlists, RuntimeConfig,
            SecurityPolicy,
        },
    },
    ::routing_test_helpers::{generate_storage_path, RoutingTestModel, RoutingTestModelBuilder},
    anyhow::anyhow,
    async_trait::async_trait,
    cm_moniker::InstancedRelativeMoniker,
    cm_rust::*,
    cm_types::Url,
    fidl::{
        self,
        endpoints::{self, create_proxy, ClientEnd, Proxy, ServerEnd},
    },
    fidl_fidl_examples_routing_echo::{self as echo},
    fidl_fuchsia_component as fcomponent, fidl_fuchsia_component_decl as fdecl,
    fidl_fuchsia_component_runner as fcrunner, fidl_fuchsia_io as fio, fidl_fuchsia_sys2 as fsys,
    fuchsia_inspect as inspect, fuchsia_zircon as zx,
    futures::lock::Mutex,
    futures::prelude::*,
    moniker::{AbsoluteMoniker, AbsoluteMonikerBase, ChildMoniker, ChildMonikerBase},
    std::{
        collections::{HashMap, HashSet},
        convert::{TryFrom, TryInto},
        default::Default,
        fs,
        path::{Path, PathBuf},
        sync::Arc,
    },
    tempfile::TempDir,
    vfs::directory::entry::DirectoryEntry,
};

// TODO(https://fxbug.dev/61861): remove type aliases once the routing_test_helpers lib has a stable
// API.
pub type ExpectedResult = ::routing_test_helpers::ExpectedResult;
pub type CheckUse = ::routing_test_helpers::CheckUse;

/// 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 BuiltinRunnerFactory>>,
    mock_builtin_runners: HashSet<CapabilityName>,
    namespace_capabilities: Vec<CapabilityDecl>,
    builtin_capabilities: Vec<CapabilityDecl>,
    component_id_index_path: Option<String>,
    // 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>,
    capability_policy: HashMap<CapabilityAllowlistKey, HashSet<AllowlistEntry>>,
    debug_capability_policy: HashMap<CapabilityAllowlistKey, HashSet<(AbsoluteMoniker, String)>>,
    child_policy: ChildPolicyAllowlists,
    reboot_on_terminate_enabled: bool,
}

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", registered in the root's environment
    /// under the given name.
    pub fn add_builtin_runner(mut self, name: &str, runner: Arc<dyn BuiltinRunnerFactory>) -> 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 fn set_namespace_capabilities(mut self, caps: Vec<CapabilityDecl>) -> Self {
        self.namespace_capabilities = caps;
        self
    }

    /// Set the list of built-in capabilities that are provided from component manager.
    /// The `test_runner` capability is always provided.
    pub fn set_builtin_capabilities(mut self, caps: Vec<CapabilityDecl>) -> Self {
        self.builtin_capabilities = caps;
        self
    }

    pub fn set_component_id_index_path(mut self, index_path: String) -> Self {
        self.component_id_index_path = Some(index_path);
        self
    }

    pub fn set_reboot_on_terminate_enabled(mut self, val: bool) -> Self {
        self.reboot_on_terminate_enabled = val;
        self
    }

    pub fn set_reboot_on_terminate_policy(mut self, allowlist: Vec<AllowlistEntry>) -> Self {
        self.child_policy.reboot_on_terminate = allowlist;
        self
    }

    /// Add a custom capability security policy to restrict routing of certain caps.
    pub fn add_capability_policy(
        mut self,
        key: CapabilityAllowlistKey,
        allowlist: HashSet<AllowlistEntry>,
    ) -> Self {
        self.capability_policy.insert(key, allowlist);
        self
    }

    /// Add a custom debug capability security policy to restrict routing of certain caps.
    pub fn add_debug_capability_policy(
        mut self,
        key: CapabilityAllowlistKey,
        allowlist: HashSet<(AbsoluteMoniker, String)>,
    ) -> Self {
        self.debug_capability_policy.insert(key, allowlist);
        self
    }

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

#[async_trait]
impl RoutingTestModelBuilder for RoutingTestBuilder {
    type Model = RoutingTest;

    fn new(root_component: &str, components: Vec<(&'static str, ComponentDecl)>) -> Self {
        Self::new(root_component, components)
    }

    fn set_namespace_capabilities(&mut self, caps: Vec<CapabilityDecl>) {
        self.namespace_capabilities = caps;
    }

    fn set_builtin_capabilities(&mut self, caps: Vec<CapabilityDecl>) {
        self.builtin_capabilities = caps;
    }

    fn register_mock_builtin_runner(&mut self, runner: &str) {
        self.mock_builtin_runners.insert(runner.into());
    }

    fn add_capability_policy(
        &mut self,
        key: CapabilityAllowlistKey,
        allowlist: HashSet<AllowlistEntry>,
    ) {
        self.capability_policy.insert(key, allowlist);
    }

    fn add_debug_capability_policy(
        &mut self,
        key: CapabilityAllowlistKey,
        allowlist: HashSet<(AbsoluteMoniker, String)>,
    ) {
        self.debug_capability_policy.insert(key, allowlist);
    }

    fn set_component_id_index_path(&mut self, index_path: String) {
        self.component_id_index_path = Some(index_path);
    }

    async fn build(self) -> RoutingTest {
        self.build().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,
    pub test_dir_proxy: fio::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 {
        let mock_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"), "hello")
            .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(),
            };
            mock_runner.add_host_fn(&format!("test:///{}_resolved", name), host_fn);
            mock_resolver.add_component(name, decl.clone());
        }

        let echo_service = Arc::new(EchoService::new());

        // Add the `test_runner` capability as a built-in.
        builder.builtin_capabilities.push(CapabilityDecl::Runner(RunnerDecl {
            name: TEST_RUNNER_NAME.into(),
            source_path: None,
        }));

        let config = RuntimeConfig {
            namespace_capabilities: builder.namespace_capabilities,
            builtin_capabilities: builder.builtin_capabilities,
            root_component_url: Some(
                Url::new(format!("test:///{}", builder.root_component)).unwrap(),
            ),
            security_policy: SecurityPolicy {
                capability_policy: builder.capability_policy,
                debug_capability_policy: builder.debug_capability_policy,
                child_policy: builder.child_policy,
                ..Default::default()
            },
            component_id_index_path: builder.component_id_index_path,
            reboot_on_terminate_enabled: builder.reboot_on_terminate_enabled,
            ..Default::default()
        };
        let inspector = inspect::Inspector::new();
        let mut env_builder = BuiltinEnvironmentBuilder::new()
            .set_inspector(inspector)
            .set_runtime_config(config)
            .add_resolver("test".to_string(), Box::new(mock_resolver))
            .add_runner(TEST_RUNNER_NAME.into(), mock_runner.clone());
        for name in builder.mock_builtin_runners.clone() {
            env_builder = env_builder.add_runner(name, mock_runner.clone())
        }
        for (name, runner) in builder.builtin_runners.clone() {
            env_builder = env_builder.add_runner(name, runner);
        }
        let builtin_environment =
            env_builder.build().await.expect("builtin environment setup failed");

        let model = builtin_environment.model.clone();
        model.root().hooks.install(builder.additional_hooks.clone()).await;
        model.root().hooks.install(echo_service.hooks()).await;

        Self {
            components: builder.components,
            model,
            builtin_environment,
            _echo_service: echo_service,
            mock_runner,
            test_dir,
            test_dir_proxy,
            root_component_name: builder.root_component.clone(),
        }
    }

    pub fn test_dir_path(&self) -> &Path {
        self.test_dir.path()
    }

    /// Set up the given OutDir, installing a set of files assumed to exist by
    /// many tests:
    ///   - A file `/svc/foo` implementing `fidl.examples.routing.echo.Echo`.
    ///   - A static file `/svc/file`, containing the string "hippos" encoded as UTF-8.
    pub 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: impl Into<ChildDecl>,
    ) {
        self.create_dynamic_child_with_args(
            moniker,
            collection,
            decl,
            fcomponent::CreateChildArgs::EMPTY,
        )
        .await
    }

    /// Creates a dynamic child `child_decl` in `moniker`'s `collection`.
    pub async fn create_dynamic_child_with_args<'a>(
        &'a self,
        moniker: AbsoluteMoniker,
        collection: &'a str,
        decl: impl Into<ChildDecl>,
        args: fcomponent::CreateChildArgs,
    ) {
        let component_name =
            self.start_instance_and_wait_start(&moniker).await.expect("start 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.into(), args).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 component = self.model.look_up(&moniker).await.expect("failed to look up component");
        self.model
            .start_instance(&component.abs_moniker, &StartReason::Eager)
            .await
            .expect("start instance failed");
        let child_moniker = ChildMoniker::new(name.to_string(), Some(collection.to_string()));
        let nf =
            component.remove_dynamic_child(&child_moniker).await.expect("failed to remove child");
        // Wait for destruction to fully complete.
        nf.await.expect("failed to destroy child");
    }

    pub async fn bind_and_get_namespace(&self, moniker: AbsoluteMoniker) -> Arc<ManagedNamespace> {
        let component_name = self.start_instance_and_wait_start(&moniker).await.unwrap();
        let component_resolved_url = Self::resolved_url(&component_name);
        let namespace = self
            .mock_runner
            .get_namespace(&component_resolved_url)
            .expect("could not find namespace");
        Self::check_namespace(component_name, &self.mock_runner, self.components.clone()).await;
        namespace
    }

    /// Lists the contents of a storage directory.
    pub async fn list_directory_in_storage(
        &self,
        subdir: Option<&str>,
        relation: InstancedRelativeMoniker,
        instance_id: Option<&ComponentInstanceId>,
        relative_path: &str,
    ) -> Vec<String> {
        let dir_path = generate_storage_path(subdir.map(|s| s.to_string()), &relation, instance_id);
        let mut dir_path = dir_path.parent().unwrap().to_path_buf();
        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
        }
    }

    /// Lists the contents of a directory.
    pub async fn list_directory(&self, path: &str) -> Vec<String> {
        let dir_proxy = io_util::open_directory(
            &self.test_dir_proxy,
            Path::new(path),
            io_util::OPEN_RIGHT_READABLE,
        )
        .expect("failed to open directory");
        list_directory(&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.dirname),
                UseDecl::Protocol(s) => Some(s.target_path.dirname),
                UseDecl::Storage(s) => Some(s.target_path.to_string()),
                UseDecl::Event(_) | UseDecl::EventStreamDeprecated(_) => None,
                UseDecl::EventStream(_) => {
                    // TODO(fxbug.dev/81980): Route EventStream path
                    None
                }
            })
            .collect();
        let mut expected_paths = vec![];
        expected_paths.extend(expected_paths_hs.into_iter());

        // The "pkg" directory has been added to the mocked ResolvedComponent.
        // We need to test that it exists.
        expected_paths.push("/pkg".to_string());

        // 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.component.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.start_instance_and_wait_start(&moniker).await.expect("bind instance failed");
        let component_resolved_url = Self::resolved_url(&component_name);
        let path = "/svc/fuchsia.component.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;
    }

    /// Build an outgoing directory for the given component.
    fn build_outgoing_dir(
        decl: &ComponentDecl,
        test_dir_proxy: &fio::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 capability in decl.capabilities.iter() {
            let path = match capability {
                CapabilityDecl::Protocol(ProtocolDecl { source_path, .. }) => source_path.as_ref(),
                CapabilityDecl::Directory(DirectoryDecl { source_path, .. }) => {
                    source_path.as_ref()
                }
                _ => None,
            };
            if let Some(path) = path {
                if outgoing_paths
                    .contains_key(&CapabilityPath::try_from(&format!("{}", path) as &str).unwrap())
                {
                    // Client installed a custom DirectoryEntry at this path, don't serve the
                    // default.
                    continue;
                }
            }
            match capability {
                CapabilityDecl::Protocol(_) => {
                    Self::install_default_out_files(&mut out_dir);
                }
                CapabilityDecl::Directory(_) => out_dir.add_directory_proxy(test_dir_proxy),
                CapabilityDecl::Storage(storage) => {
                    // 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
    }

    /// Attempt to start the instance associated with the given moniker with the
    /// default reason of StartReason::Eager.
    ///
    /// On success, returns the short name of the component.
    pub async fn start_instance(&self, moniker: &AbsoluteMoniker) -> Result<String, ModelError> {
        self.start_instance_with(moniker, StartReason::Eager, false).await
    }

    /// Attempt to start the instance associated with the given moniker with the default reason of
    /// StartReason::Eager, and waits for the runner to start the component. This method will only
    /// work if the component in question is using the default mock runner (otherwise, it will
    /// hang).
    ///
    /// On success, returns the short name of the component.
    pub async fn start_instance_and_wait_start(
        &self,
        moniker: &AbsoluteMoniker,
    ) -> Result<String, ModelError> {
        self.start_instance_with(moniker, StartReason::Eager, true).await
    }

    async fn start_instance_with(
        &self,
        moniker: &AbsoluteMoniker,
        reason: StartReason,
        wait_for_start: bool,
    ) -> Result<String, ModelError> {
        self.model.start_instance(moniker, &reason).await?;
        let component_name = match moniker.path().last() {
            Some(part) => part.name().to_string(),
            None => self.root_component_name.to_string(),
        };
        if wait_for_start {
            let resolved_url = Self::resolved_url(&component_name);
            self.mock_runner.wait_for_url(&resolved_url).await;
        }
        Ok(component_name)
    }

    pub async fn start_and_get_instance<'a>(
        &self,
        moniker: &AbsoluteMoniker,
        reason: StartReason,
        wait_for_start: bool,
    ) -> Result<Arc<ComponentInstance>, ModelError> {
        let instance = self.model.start_instance(moniker, &reason).await?;
        let component_name = match moniker.path().last() {
            Some(part) => part.name().to_string(),
            None => self.root_component_name.to_string(),
        };
        if wait_for_start {
            let resolved_url = Self::resolved_url(&component_name);
            self.mock_runner.wait_for_url(&resolved_url).await;
        }

        Ok(instance)
    }

    /// 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, moniker: &AbsoluteMoniker) {
        // Lookup, start, and open a connection to the component's outgoing directory.
        let (dir_proxy, server_end) =
            fidl::endpoints::create_proxy::<fio::DirectoryMarker>().unwrap();
        self.model.look_up(moniker).await.expect("lookup component failed");
        let mut server_end = server_end.into_channel();
        self.model
            .start_instance(moniker, &StartReason::Eager)
            .await
            .expect("failed to start component")
            .open_outgoing(
                fio::OPEN_RIGHT_READABLE,
                fio::MODE_TYPE_DIRECTORY,
                PathBuf::from("/."),
                &mut server_end,
            )
            .await
            .expect("failed to open component's outgoing directory");

        // Ensure we can successfully talk to the directory.
        dir_proxy
            .sync()
            .await
            .expect("could not communicate with directory")
            .map_err(zx::Status::from_raw)
            .expect("failed to sync directory");
    }

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

#[async_trait]
impl RoutingTestModel for RoutingTest {
    type C = ComponentInstance;

    async fn check_use(&self, moniker: AbsoluteMoniker, check: CheckUse) {
        let component_name = self
            .start_instance_and_wait_start(&moniker)
            .await
            .expect(&format!("start instance failed for `{}`", moniker));
        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, expected_res } => {
                capability_util::call_echo_svc_from_namespace(&namespace, path, expected_res).await;
            }
            CheckUse::Service { path, instance, member, expected_res } => {
                capability_util::call_service_instance_echo_svc_from_namespace(
                    &namespace,
                    path,
                    instance,
                    member,
                    expected_res,
                )
                .await;
            }
            CheckUse::Directory { path, file, expected_res } => {
                capability_util::read_data_from_namespace(&namespace, path, &file, expected_res)
                    .await
            }
            CheckUse::Storage {
                path,
                storage_relation,
                from_cm_namespace,
                storage_subdir,
                expected_res,
            } => {
                if let ExpectedResult::Ok = &expected_res {
                    assert!(
                        storage_relation.is_some(),
                        "relative moniker required if expected result is ok"
                    );
                }
                capability_util::write_file_to_storage(&namespace, path, expected_res.clone())
                    .await;

                let instance_id = self
                    .model
                    .root()
                    .try_get_component_id_index()
                    .unwrap()
                    .look_up_moniker(&moniker)
                    .cloned();

                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");
                        let res = capability_util::check_file_in_storage(
                            storage_subdir,
                            relative_moniker,
                            instance_id.as_ref(),
                            &tmp_proxy,
                        )
                        .await;
                        if let ExpectedResult::Ok = &expected_res {
                            res.expect("failed to read file");
                        }
                    } else {
                        // Check for the file in the test's isolated test directory
                        let res = capability_util::check_file_in_storage(
                            storage_subdir,
                            relative_moniker,
                            instance_id.as_ref(),
                            &self.test_dir_proxy,
                        )
                        .await;
                        if let ExpectedResult::Ok = &expected_res {
                            res.expect("failed to read file");
                        }
                    }
                }
            }
            CheckUse::StorageAdmin {
                storage_relation,
                from_cm_namespace,
                storage_subdir,
                expected_res,
            } => {
                let storage_admin_proxy =
                    capability_util::connect_to_svc_in_namespace::<fsys::StorageAdminMarker>(
                        &namespace,
                        &"/svc/fuchsia.sys2.StorageAdmin".try_into().unwrap(),
                    )
                    .await;
                let (storage_proxy, server_end) = create_proxy().unwrap();
                let flags = fio::OPEN_RIGHT_WRITABLE | fio::OPEN_FLAG_CREATE;
                let relative_moniker_string = format!("{}", storage_relation);
                let component_abs_moniker = AbsoluteMoniker::from_relative(
                    &moniker,
                    &storage_relation.to_relative_moniker(),
                )
                .unwrap();
                let component_instance_id = self
                    .model
                    .root()
                    .try_get_component_id_index()
                    .unwrap()
                    .look_up_moniker(&component_abs_moniker)
                    .cloned();
                storage_admin_proxy
                    .open_component_storage(
                        relative_moniker_string.as_str(),
                        flags,
                        fio::MODE_TYPE_DIRECTORY,
                        server_end,
                    )
                    .expect("failed to open component storage");

                let storage_proxy =
                    fio::DirectoryProxy::from_channel(storage_proxy.into_channel().unwrap());

                capability_util::write_hippo_file_to_directory(
                    &storage_proxy,
                    expected_res.clone(),
                )
                .await;
                if expected_res == ExpectedResult::Ok {
                    let storage_dir = if from_cm_namespace {
                        io_util::open_directory_in_namespace("/tmp", io_util::OPEN_RIGHT_READABLE)
                            .expect("failed to open /tmp")
                    } else {
                        io_util::clone_directory(&self.test_dir_proxy, fio::CLONE_FLAG_SAME_RIGHTS)
                            .expect("failed to clone test_dir_proxy")
                    };
                    capability_util::check_file_in_storage(
                        storage_subdir.clone(),
                        storage_relation.clone(),
                        component_instance_id.as_ref(),
                        &storage_dir,
                    )
                    .await
                    .expect("failed to read file");
                    storage_admin_proxy
                        .delete_component_storage(relative_moniker_string.as_str())
                        .await
                        .expect("failed to send fidl message")
                        .expect("error encountered while deleting component storage");
                    capability_util::confirm_storage_is_deleted_for_component(
                        storage_subdir,
                        storage_relation,
                        component_instance_id.as_ref(),
                        &storage_dir,
                    )
                    .await;
                }
            }
            CheckUse::Event { request, expected_res } => {
                // Fails if the component did not use the protocol EventSource or if the event is
                // not allowed.
                capability_util::subscribe_to_event_stream(&namespace, expected_res, request).await;
            }
        }
    }

    async fn check_use_exposed_dir(&self, moniker: AbsoluteMoniker, check: CheckUse) {
        match check {
            CheckUse::Protocol { path, expected_res } => {
                capability_util::call_echo_svc_from_exposed_dir(
                    path,
                    &moniker,
                    &self.model,
                    expected_res,
                )
                .await;
            }
            CheckUse::Service { path, instance, member, expected_res } => {
                capability_util::call_service_instance_echo_svc_from_exposed_dir(
                    path,
                    instance,
                    member,
                    &moniker,
                    &self.model,
                    expected_res,
                )
                .await;
            }
            CheckUse::Directory { path, file, expected_res } => {
                capability_util::read_data_from_exposed_dir(
                    path,
                    &file,
                    &moniker,
                    &self.model,
                    expected_res,
                )
                .await;
            }
            CheckUse::Storage { .. } => {
                panic!("storage capabilities can't be exposed");
            }
            CheckUse::StorageAdmin { .. } => {
                panic!("unimplemented");
            }
            CheckUse::Event { .. } => {
                panic!("event capabilities can't be exposed");
            }
        }
    }

    async fn look_up_instance(
        &self,
        moniker: &AbsoluteMoniker,
    ) -> Result<Arc<ComponentInstance>, anyhow::Error> {
        self.model.look_up(&moniker).await.map_err(|err| anyhow!(err))
    }

    async fn check_open_file(&self, moniker: AbsoluteMoniker, path: CapabilityPath) {
        let component_name =
            self.start_instance_and_wait_start(&moniker).await.expect("start 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;
    }

    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
    }

    fn install_namespace_directory(&self, path: &str) {
        let (client_chan, server_chan) = zx::Channel::create().unwrap();
        let ns = fdio::Namespace::installed().expect("Failed to get installed namespace");
        ns.bind(path, client_chan).expect(&format!("Failed to bind dir {}", path));
        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));
    }

    fn add_subdir_to_data_directory(&self, subdir: &str) {
        fs::create_dir_all(self.test_dir.path().join(subdir)).unwrap()
    }

    async fn check_test_subdir_contents(&self, path: &str, expected: Vec<String>) {
        assert_eq!(self.list_directory(path).await, expected)
    }

    async fn check_namespace_subdir_contents(&self, path: &str, expected: Vec<String>) {
        let dir_proxy = io_util::open_directory_in_namespace(path, io_util::OPEN_RIGHT_READABLE)
            .expect("failed to open directory");
        assert_eq!(list_directory(&dir_proxy).await, expected)
    }

    async fn check_test_subdir_contains(&self, path: &str, expected: String) {
        assert!(self.list_directory(path).await.contains(&expected))
    }

    async fn check_test_dir_tree_contains(&self, expected: String) {
        assert!(list_directory_recursive(&self.test_dir_proxy)
            .await
            .iter()
            .find(|&name| name.starts_with(&expected))
            .is_some());
    }
}

/// Installs a new directory at `path` in the test's namespace, removing it when this object
/// goes out of scope.
pub struct ScopedNamespaceDir<'a> {
    path: &'a str,
}

impl<'a> ScopedNamespaceDir<'a> {
    pub fn new(test: &RoutingTest, path: &'a str) -> Self {
        test.install_namespace_directory(path);
        Self { path }
    }
}

impl Drop for ScopedNamespaceDir<'_> {
    fn drop(&mut self) {
        let ns = fdio::Namespace::installed().expect("Failed to get installed namespace");
        ns.unbind(self.path).expect(&format!("Failed to unbind dir {}", self.path));
    }
}

/// Contains functions to use capabilities in routing tests.
pub mod capability_util {
    use {
        super::*, anyhow::format_err, assert_matches::assert_matches, cm_rust::NativeIntoFidl,
        fidl::endpoints::ProtocolMarker, fidl_fuchsia_sys2::EventSourceMarker, std::path::PathBuf,
    };

    /// Looks up `resolved_url` in the namespace, and attempts to read ${path}/hippo. The file
    /// should contain the string "hello".
    pub async fn read_data_from_namespace(
        namespace: &ManagedNamespace,
        path: CapabilityPath,
        file: &Path,
        expected_res: ExpectedResult,
    ) {
        let path = path.to_string();
        let dir_proxy = take_dir_from_namespace(namespace, &path).await;
        let file_proxy = io_util::open_file(&dir_proxy, file, fio::OPEN_RIGHT_READABLE)
            .expect("failed to open file");
        let res = io_util::read_file(&file_proxy).await;
        match expected_res {
            ExpectedResult::Ok => assert_eq!(
                "hello",
                res.expect(&format!("failed to read file {}", path.to_string()))
            ),
            ExpectedResult::Err(s) => {
                assert!(res.is_err(), "read file successfully when it should fail");
                let epitaph = dir_proxy.take_event_stream().next().await.expect("no epitaph");
                assert_matches!(
                    epitaph,
                    Err(fidl::Error::ClientChannelClosed {
                        status, ..
                    }) if status == s
                );
            }
            ExpectedResult::ErrWithNoEpitaph => {
                assert!(res.is_err(), "read file successfully when it should fail");
                assert_matches!(dir_proxy.take_event_stream().next().await, None);
            }
        }
        // We took ownership of `dir_proxy`, add it back to the namespace.
        add_dir_to_namespace(namespace, &path, dir_proxy).await;
    }

    pub async fn write_file_to_storage(
        namespace: &ManagedNamespace,
        path: CapabilityPath,
        expected_res: ExpectedResult,
    ) {
        let dir_path = path.to_string();
        let dir_proxy = take_dir_from_namespace(namespace, dir_path.as_str()).await;
        write_hippo_file_to_directory(&dir_proxy, expected_res).await;
        add_dir_to_namespace(namespace, dir_path.as_str(), dir_proxy).await;
    }

    pub async fn write_hippo_file_to_directory(
        dir_proxy: &fio::DirectoryProxy,
        expected_res: ExpectedResult,
    ) {
        let (file_proxy, server_end) = create_proxy::<fio::FileMarker>().unwrap();
        let flags = fio::OPEN_RIGHT_WRITABLE | fio::OPEN_FLAG_CREATE;
        let res = async {
            dir_proxy.open(
                flags,
                fio::MODE_TYPE_FILE,
                "hippos",
                ServerEnd::new(server_end.into_channel()),
            )?;
            file_proxy.write(b"hippos can be stored here").await
        }
        .await;
        match expected_res {
            ExpectedResult::Ok => {
                let _: u64 = res
                    .expect("failed to write file")
                    .map_err(zx::Status::from_raw)
                    .expect("write error");
            }
            ExpectedResult::Err(s) => {
                res.expect_err("unexpectedly succeeded writing file");
                let epitaph = dir_proxy.take_event_stream().next().await.expect("no epitaph");
                assert_matches!(
                    epitaph,
                    Err(fidl::Error::ClientChannelClosed {
                        status, ..
                    }) if status == s
                );
            }
            ExpectedResult::ErrWithNoEpitaph => {
                res.expect_err("unexpectedly succeeded writing file");
                assert_matches!(dir_proxy.take_event_stream().next().await, None);
            }
        }
    }

    /// Create a file with the given contents in the test dir, along with any subdirectories
    /// required.
    pub(super) async fn create_static_file(
        root: &fio::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()),
                fio::OPEN_RIGHT_WRITABLE | fio::OPEN_FLAG_CREATE,
            )?
        } else {
            io_util::open_file(root, path, fio::OPEN_RIGHT_WRITABLE | fio::OPEN_FLAG_CREATE)?
        };

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

    pub async fn check_file_in_storage(
        storage_subdir: Option<String>,
        relation: InstancedRelativeMoniker,
        instance_id: Option<&ComponentInstanceId>,
        test_dir_proxy: &fio::DirectoryProxy,
    ) -> Result<(), anyhow::Error> {
        let mut dir_path = generate_storage_path(storage_subdir, &relation, instance_id);
        dir_path.push("hippos");
        let file_proxy =
            io_util::open_file(&test_dir_proxy, &dir_path, io_util::OPEN_RIGHT_READABLE)?;
        let res = io_util::read_file(&file_proxy).await;

        if let Ok(contents) = res {
            assert_eq!("hippos can be stored here".to_string(), contents);
            Ok(())
        } else {
            Err(res.expect_err("failed to read file"))
        }
    }

    pub async fn confirm_storage_is_deleted_for_component(
        storage_subdir: Option<String>,
        relation: InstancedRelativeMoniker,
        instance_id: Option<&ComponentInstanceId>,
        test_dir_proxy: &fio::DirectoryProxy,
    ) {
        let dir_path = generate_storage_path(storage_subdir, &relation, instance_id);
        let res = io_util::directory::open_directory(
            &test_dir_proxy,
            dir_path.to_str().unwrap(),
            io_util::OPEN_RIGHT_READABLE,
        )
        .await
        .expect_err("open_directory shouldnt have succeeded");
        assert_eq!(
            format!("{:?}", res),
            format!("{:?}", io_util::node::OpenError::OpenError(zx::Status::NOT_FOUND))
        );
    }

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

    pub async fn connect_to_instance_svc_in_namespace<T: ProtocolMarker>(
        namespace: &ManagedNamespace,
        path: &CapabilityPath,
        instance: &str,
        member: &str,
    ) -> T::Proxy {
        let dir_proxy = take_dir_from_namespace(namespace, &path.dirname).await;
        let service_dir = io_util::directory::open_directory(
            &dir_proxy,
            &path.basename,
            fio::OPEN_RIGHT_READABLE | fio::OPEN_RIGHT_WRITABLE,
        )
        .await
        .expect("failed to open service dir");
        let instance_dir = io_util::directory::open_directory(
            &service_dir,
            instance,
            fio::OPEN_RIGHT_READABLE | fio::OPEN_RIGHT_WRITABLE,
        )
        .await
        .expect("failed to open instance dir");
        let member_proxy = io_util::directory::open_node_no_describe(
            &instance_dir,
            member,
            fio::OPEN_RIGHT_READABLE | fio::OPEN_RIGHT_WRITABLE,
            fio::MODE_TYPE_SERVICE,
        )
        .expect("failed to open member node");
        add_dir_to_namespace(namespace, &path.dirname, dir_proxy).await;
        let client_end =
            ClientEnd::<T>::new(member_proxy.into_channel().unwrap().into_zx_channel());
        client_end.into_proxy().unwrap()
    }

    /// Verifies that it's possible to subscribe to the given `event` by connecting to an
    /// `EventSource` on the given `namespace`. Used to test event capability 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,
        expected_res: ExpectedResult,
        event: ::routing::event::EventSubscription,
    ) {
        let res = subscribe_to_event(namespace, event).await;
        match (res, expected_res) {
            (Err(e), ExpectedResult::Ok) => {
                panic!("unexpected failure {}", e);
            }
            (Ok(_), ExpectedResult::Err(_)) | (Ok(_), ExpectedResult::ErrWithNoEpitaph) => {
                panic!("unexpected success");
            }
            _ => {}
        }
    }

    pub async fn subscribe_to_event(
        namespace: &ManagedNamespace,
        event: ::routing::event::EventSubscription,
    ) -> Result<fsys::EventStreamRequestStream, anyhow::Error> {
        let event_source_path = CapabilityPath::try_from("/svc/fuchsia.sys2.EventSource").unwrap();
        let event_source_proxy =
            connect_to_svc_in_namespace::<EventSourceMarker>(namespace, &event_source_path).await;
        let (client_end, stream) =
            fidl::endpoints::create_request_stream::<fsys::EventStreamMarker>()?;
        // Bind the future to a variable in order to avoid using `event` across an await.
        let subscribe_future = event_source_proxy.subscribe(
            &mut vec![fsys::EventSubscription {
                event_name: Some(event.event_name.to_string()),
                ..fsys::EventSubscription::EMPTY
            }]
            .into_iter(),
            client_end,
        );
        subscribe_future
            .await?
            .map_err(|error| format_err!("Unable to subscribe to event stream: {:?}", error))?;
        Ok(stream)
    }

    /// Looks up `resolved_url` in the namespace, and attempts to use `path`. Expects the service
    /// to be fidl.examples.routing.echo.Echo.
    pub async fn call_echo_svc_from_namespace(
        namespace: &ManagedNamespace,
        path: CapabilityPath,
        expected_res: ExpectedResult,
    ) {
        let echo_proxy = connect_to_svc_in_namespace::<echo::EchoMarker>(namespace, &path).await;
        call_echo_and_validate_result(echo_proxy, expected_res).await;
    }

    /// Looks up `resolved_url` in the namespace, and attempts to use `instance` at `path`. Expects the service
    /// to be fidl.examples.routing.echo.Echo.
    pub async fn call_service_instance_echo_svc_from_namespace(
        namespace: &ManagedNamespace,
        path: CapabilityPath,
        instance: String,
        member: String,
        expected_res: ExpectedResult,
    ) {
        let echo_proxy = connect_to_instance_svc_in_namespace::<echo::EchoMarker>(
            namespace, &path, &instance, &member,
        )
        .await;
        call_echo_and_validate_result(echo_proxy, expected_res).await;
    }

    async fn call_echo_and_validate_result(
        echo_proxy: echo::EchoProxy,
        expected_res: ExpectedResult,
    ) {
        let res = echo_proxy.echo_string(Some("hippos")).await;
        match expected_res {
            ExpectedResult::Ok => {
                assert_eq!(res.expect("failed to use echo service"), Some("hippos".to_string()))
            }
            ExpectedResult::Err(s) => {
                let err = res.expect_err("used echo service successfully when it should fail");
                assert!(err.is_closed(), "expected file closed error, got: {:?}", err);
                let epitaph = echo_proxy.take_event_stream().next().await.expect("no epitaph");
                assert_matches!(
                    epitaph,
                    Err(fidl::Error::ClientChannelClosed {
                        status, ..
                    }) if status == s
                );
            }
            ExpectedResult::ErrWithNoEpitaph => {
                let err = res.expect_err("used echo service successfully when it should fail");
                assert!(err.is_closed(), "expected file closed error, got: {:?}", err);
                assert_matches!(echo_proxy.take_event_stream().next().await, None);
            }
        }
    }

    /// 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 = take_dir_from_namespace(namespace, &path.dirname).await;
        let node_proxy = io_util::open_node(
            &dir_proxy,
            &Path::new(&path.basename),
            fio::OPEN_RIGHT_READABLE | fio::OPEN_FLAG_DESCRIBE,
            // This should be MODE_TYPE_SERVICE, but we implement the underlying
            // service as a file for convenience in testing.
            fio::MODE_TYPE_FILE,
        )
        .expect("failed to open file service");
        add_dir_to_namespace(namespace, &path.dirname, dir_proxy).await;

        let file_proxy = fio::FileProxy::new(node_proxy.into_channel().unwrap());
        let mut event_stream = file_proxy.take_event_stream();
        let event = event_stream.try_next().await.unwrap();
        match event.expect("failed to received file event") {
            fio::FileEvent::OnOpen_ { s, info } => {
                assert_eq!(s, zx::sys::ZX_OK);
                assert!(matches!(
                    *info.expect("failed to receive node info"),
                    fio::NodeInfo::Vmofile(fio::Vmofile { .. })
                ));
            }
            fio::FileEvent::OnConnectionInfo { info } => {
                assert!(matches!(
                    info.representation.expect("failed to receive node info"),
                    fio::Representation::Memory(fio::MemoryInfo { .. })
                ));
            }
        }
    }

    /// Attempts to read ${path}/hippo in `abs_moniker`'s exposed directory. The file should
    /// contain the string "hello".
    pub async fn read_data_from_exposed_dir<'a>(
        path: CapabilityPath,
        file: &Path,
        abs_moniker: &'a AbsoluteMoniker,
        model: &'a Arc<Model>,
        expected_res: ExpectedResult,
    ) {
        let (node_proxy, server_end) = endpoints::create_proxy::<fio::NodeMarker>().unwrap();
        open_exposed_dir(&path, abs_moniker, model, fio::MODE_TYPE_DIRECTORY, server_end).await;
        let dir_proxy = fio::DirectoryProxy::new(node_proxy.into_channel().unwrap());
        match expected_res {
            ExpectedResult::Ok => {
                let file_proxy = io_util::open_file(&dir_proxy, &file, fio::OPEN_RIGHT_READABLE)
                    .expect("failed to open file");
                let res = io_util::read_file(&file_proxy).await;
                assert_eq!("hello", res.expect("failed to read file"));
            }
            ExpectedResult::Err(s) => {
                io_util::open_file(&dir_proxy, &file, fio::OPEN_RIGHT_READABLE)
                    .expect_err("opened file successfully when it should fail");
                let epitaph = dir_proxy.take_event_stream().next().await.expect("no epitaph");
                assert_matches!(
                    epitaph,
                    Err(fidl::Error::ClientChannelClosed {
                        status, ..
                    }) if status == s
                );
            }
            ExpectedResult::ErrWithNoEpitaph => {
                io_util::open_file(&dir_proxy, &file, fio::OPEN_RIGHT_READABLE)
                    .expect_err("opened file successfully when it should fail");
                assert_matches!(dir_proxy.take_event_stream().next().await, None);
            }
        }
    }

    /// Attempts to use the fidl.examples.routing.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>,
        expected_res: ExpectedResult,
    ) {
        let (node_proxy, server_end) = endpoints::create_proxy::<fio::NodeMarker>().unwrap();
        open_exposed_dir(&path, abs_moniker, model, fio::MODE_TYPE_SERVICE, server_end).await;
        let echo_proxy = echo::EchoProxy::new(node_proxy.into_channel().unwrap());
        call_echo_and_validate_result(echo_proxy, expected_res).await;
    }

    pub async fn call_service_instance_echo_svc_from_exposed_dir(
        path: CapabilityPath,
        instance: String,
        member: String,
        abs_moniker: &AbsoluteMoniker,
        model: &Arc<Model>,
        expected_res: ExpectedResult,
    ) {
        let (node_proxy, server_end) = endpoints::create_proxy::<fio::NodeMarker>().unwrap();
        open_exposed_dir(&path, abs_moniker, model, fio::MODE_TYPE_SERVICE, server_end).await;
        let service_dir = fio::DirectoryProxy::from_channel(node_proxy.into_channel().unwrap());
        let instance_dir = io_util::directory::open_directory(
            &service_dir,
            &instance,
            io_util::OPEN_RIGHT_READABLE | io_util::OPEN_RIGHT_WRITABLE,
        )
        .await
        .expect("failed to open instance");
        let member_node = io_util::directory::open_node_no_describe(
            &instance_dir,
            &member,
            io_util::OPEN_RIGHT_READABLE | io_util::OPEN_RIGHT_WRITABLE,
            fio::MODE_TYPE_SERVICE,
        )
        .expect("failed to open member node");
        let echo_proxy = echo::EchoProxy::new(member_node.into_channel().unwrap());
        call_echo_and_validate_result(echo_proxy, expected_res).await;
    }

    /// Looks up `resolved_url` in the namespace, and attempts to use `path`. Expects the service
    /// to be fuchsia.component.Realm.
    pub async fn call_realm_svc(
        path: CapabilityPath,
        resolved_url: &str,
        namespace: &ManagedNamespace,
        bind_calls: Arc<Mutex<Vec<String>>>,
    ) {
        let dir_proxy = take_dir_from_namespace(namespace, &path.dirname).await;
        let node_proxy = io_util::open_node(
            &dir_proxy,
            &Path::new(&path.basename),
            fio::OPEN_RIGHT_READABLE | fio::OPEN_RIGHT_WRITABLE,
            fio::MODE_TYPE_SERVICE,
        )
        .expect("failed to open realm service");
        let realm_proxy = fcomponent::RealmProxy::new(node_proxy.into_channel().unwrap());
        let mut child_ref = fdecl::ChildRef { name: "my_child".to_string(), collection: None };
        let (_, server_end) = fidl::endpoints::create_proxy::<fio::DirectoryMarker>().unwrap();
        let res = realm_proxy.open_exposed_dir(&mut child_ref, server_end).await;
        // Check for side effects: realm service should have received the `open_exposed_dir` call.
        res.expect("failed to send fidl message").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.component.Realm.CreateChild` to create a dynamic child.
    pub async fn call_create_child<'a>(
        namespace: &ManagedNamespace,
        collection: &'a str,
        child_decl: ChildDecl,
        args: fcomponent::CreateChildArgs,
    ) {
        let path: CapabilityPath =
            "/svc/fuchsia.component.Realm".try_into().expect("no realm service");
        let dir_proxy = take_dir_from_namespace(namespace, &path.dirname).await;
        let node_proxy = io_util::open_node(
            &dir_proxy,
            &Path::new(&path.basename),
            fio::OPEN_RIGHT_READABLE | fio::OPEN_RIGHT_WRITABLE,
            fio::MODE_TYPE_SERVICE,
        )
        .expect("failed to open realm service");
        add_dir_to_namespace(namespace, &path.dirname, dir_proxy).await;
        let realm_proxy = fcomponent::RealmProxy::new(node_proxy.into_channel().unwrap());
        let mut collection_ref = fdecl::CollectionRef { name: collection.to_string() };
        let child_decl = child_decl.native_into_fidl();
        let res = realm_proxy.create_child(&mut collection_ref, child_decl, args).await;
        res.expect("failed to send fidl message").expect("failed to create child");
    }

    /// Call `fuchsia.component.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.component.Realm".try_into().expect("no realm service");
        let dir_proxy = take_dir_from_namespace(namespace, &path.dirname).await;
        let node_proxy = io_util::open_node(
            &dir_proxy,
            &Path::new(&path.basename),
            fio::OPEN_RIGHT_READABLE | fio::OPEN_RIGHT_WRITABLE,
            fio::MODE_TYPE_SERVICE,
        )
        .expect("failed to open realm service");
        add_dir_to_namespace(namespace, &path.dirname, dir_proxy).await;
        let realm_proxy = fcomponent::RealmProxy::new(node_proxy.into_channel().unwrap());
        let mut child_ref =
            fdecl::ChildRef { collection: Some(collection.to_string()), name: name.to_string() };
        let res = realm_proxy.destroy_child(&mut child_ref).await;
        res.expect("failed to send fidl message").expect("failed to destroy child");
    }

    pub async fn take_dir_from_namespace(
        namespace: &ManagedNamespace,
        dir_path: &str,
    ) -> fio::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_path)
            .expect(&format!("didn't find dir {}", dir_path));
        let entry = ns.remove(index);
        let dir_proxy = entry.directory.unwrap().into_proxy().unwrap();
        dir_proxy
    }

    /// Adds `dir_proxy` back to the namespace. Useful for restoring the namespace after a call
    /// to `take_dir_from_namespace`.
    pub async fn add_dir_to_namespace(
        namespace: &ManagedNamespace,
        dir_path: &str,
        dir_proxy: fio::DirectoryProxy,
    ) {
        let mut ns = namespace.lock().await;
        ns.push(fcrunner::ComponentNamespaceEntry {
            path: Some(dir_path.to_string()),
            directory: Some(ClientEnd::new(dir_proxy.into_channel().unwrap().into_zx_channel())),
            ..fcrunner::ComponentNamespaceEntry::EMPTY
        });
    }

    /// 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<fio::NodeMarker>,
    ) {
        let component = model
            .look_up(abs_moniker)
            .await
            .expect(&format!("component not found {}", abs_moniker));
        model
            .start_instance(abs_moniker, &StartReason::Eager)
            .await
            .expect("failed to start instance");
        let state = component.lock_state().await;
        match &*state {
            InstanceState::Resolved(resolved_instance_state) => {
                let flags = fio::OPEN_RIGHT_READABLE | fio::OPEN_RIGHT_WRITABLE;
                let vns_path = to_fvfs_path(path);
                resolved_instance_state
                    .get_exposed_dir()
                    .open(flags, open_mode, vns_path, server_end);
            }
            _ => {
                panic!("Attempted to open exposed dir of unresolved component: {}", abs_moniker);
            }
        }
    }

    /// 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")
    }
}