flowcore 1.0.0

use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::mem::take;

use error_chain::bail;
use log::{debug, error, trace};
use serde_derive::{Deserialize, Serialize};
use url::Url;

use crate::errors::{Result, ResultExt};
use crate::model::connection::Connection;
use crate::model::connection::Direction;
use crate::model::connection::Direction::FROM;
use crate::model::connection::Direction::TO;
use crate::model::input::InputInitializer;
use crate::model::io::Find;
use crate::model::io::IOSet;
use crate::model::io::{IOType, IO};
use crate::model::metadata::MetaData;
use crate::model::name::HasName;
use crate::model::name::Name;
use crate::model::process::Process;
use crate::model::process::Process::FlowProcess;
use crate::model::process::Process::FunctionProcess;
use crate::model::process_reference::ProcessReference;
use crate::model::route::HasRoute;
use crate::model::route::SetIORoutes;
use crate::model::route::SetRoute;
use crate::model::route::{Route, RouteType};
use crate::model::validation::Validate;

/// `FlowDefinition` defines (at compile time) a parent or child flow in the nested flow hierarchy
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(deny_unknown_fields)]
pub struct FlowDefinition {
    /// `name` given to this flow
    #[serde(rename = "flow")]
    pub name: Name,
    /// `inputs` that this flow defines
    #[serde(default, rename = "input")]
    pub inputs: IOSet,
    /// `outputs` that this flow defines
    #[serde(default, rename = "output")]
    pub outputs: IOSet,
    /// Set of sub-processes referenced (used) in this flow
    #[serde(default, rename = "process")]
    pub process_refs: Vec<ProcessReference>,
    /// `connections` within this flow, from flow input or to flow outputs
    #[serde(default, rename = "connection")]
    pub connections: Vec<Connection>,
    /// `metadata` about flow author, versions etc
    #[serde(default)]
    pub metadata: MetaData,
    /// Name of any docs file associated with this Flow
    #[serde(default)]
    pub docs: String,
    /// Optional description of what this flow does
    #[serde(default)]
    pub description: String,

    /// When the same process is used multiple times within a single flow, to disambiguate
    /// between them each one must be given an alias that is used to refer to it
    #[serde(skip)]
    pub alias: Name,
    /// flows are assigned a numeric `id` in the hierarchy
    #[serde(skip)]
    pub id: usize,
    /// `source_url` is the url of the file/resource where this flow definition was read from
    #[serde(skip, default = "FlowDefinition::default_url")]
    pub source_url: Url,
    /// `route` defines the location in the hierarchy of flows where this flow resides
    #[serde(skip)]
    pub route: Route,
    /// `subprocesses` are the loaded definition of the processes reference (used) within this flow
    #[serde(skip)]
    pub subprocesses: BTreeMap<Name, Process>,
    /// `lib_references` is the set of library references used in this flow
    #[serde(skip)]
    pub lib_references: BTreeSet<Url>,
    /// `context_references` is the set of context functions used in this flow
    #[serde(skip)]
    pub context_references: BTreeSet<Url>,
}

impl Validate for FlowDefinition {
    // check the correctness of all the fields in this flow, prior to loading sub-elements
    fn validate(&self) -> Result<()> {
        for input in &self.inputs {
            input.validate()?;
        }

        for output in &self.outputs {
            output.validate()?;
        }

        for connection in &self.connections {
            connection.validate()?;
        }

        Ok(())
    }
}

impl fmt::Display for FlowDefinition {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        writeln!(f, "\tname: \t\t\t{}\n\tid: \t\t\t{}\n\talias: \t\t\t{}\n\tsource_url: \t{}\n\troute: \t\t\t{}",
                 self.name, self.id, self.alias, self.source_url, self.route)?;

        writeln!(f, "\tinputs:")?;
        for input in &self.inputs {
            writeln!(f, "\t\t\t\t\t{input:#?}")?;
        }

        writeln!(f, "\toutputs:")?;
        for output in &self.outputs {
            writeln!(f, "\t\t\t\t\t{output:#?}")?;
        }

        writeln!(f, "\tprocesses:")?;
        for flow_ref in &self.process_refs {
            writeln!(f, "\t{flow_ref}")?;
        }

        writeln!(f, "\tconnections:")?;
        for connection in &self.connections {
            writeln!(f, "\t\t\t\t\t{connection}")?;
        }

        Ok(())
    }
}

impl Default for FlowDefinition {
    fn default() -> FlowDefinition {
        FlowDefinition {
            name: String::default(),
            inputs: vec![],
            outputs: vec![],
            process_refs: vec![],
            connections: vec![],
            metadata: MetaData::default(),
            docs: String::new(),
            description: String::new(),
            alias: String::default(),
            id: 0,
            #[allow(clippy::expect_used)]
            source_url: Url::parse("file://").expect("Could not create Url"),
            route: Route::default(),
            subprocesses: BTreeMap::default(),
            lib_references: BTreeSet::default(),
            context_references: BTreeSet::default(),
        }
    }
}

impl HasName for FlowDefinition {
    fn name(&self) -> &Name {
        &self.name
    }

    fn alias(&self) -> &Name {
        &self.alias
    }
}

impl HasRoute for FlowDefinition {
    fn route(&self) -> &Route {
        &self.route
    }
    fn route_mut(&mut self) -> &mut Route {
        &mut self.route
    }
}

impl SetRoute for FlowDefinition {
    fn set_routes_from_parent(&mut self, parent_route: &Route) {
        if parent_route.is_empty() {
            self.route = Route::from(format!("/{}", self.alias));
        } else {
            self.route = Route::from(format!("{parent_route}/{}", self.alias));
        }
        self.inputs
            .set_io_routes_from_parent(&self.route, IOType::FlowInput);
        self.outputs
            .set_io_routes_from_parent(&self.route, IOType::FlowOutput);
    }
}

impl FlowDefinition {
    /// Return a default value for a Url as part of a flow
    ///
    /// # Panics
    ///
    /// This function would panic if "file://" ceased to be a valid Url, so it should never occur.
    #[must_use]
    #[allow(clippy::expect_used)]
    pub fn default_url() -> Url {
        Url::parse("file://").expect("Could not create default_url")
    }

    /// Set the alias of this flow to the supplied Name
    pub fn set_alias(&mut self, alias: &Name) {
        if alias.is_empty() {
            self.alias.clone_from(&self.name);
        } else {
            self.alias.clone_from(alias);
        }
    }

    /// Get the name of any associated docs file
    #[must_use]
    pub fn get_docs(&self) -> &str {
        &self.docs
    }

    /// Get a reference to the set of inputs this flow defines
    #[must_use]
    pub fn inputs(&self) -> &IOSet {
        &self.inputs
    }

    /// Get a mutable reference to the set of inputs this flow defines
    pub fn inputs_mut(&mut self) -> &mut IOSet {
        &mut self.inputs
    }

    /// Get a reference to the set of outputs this flow defines
    #[must_use]
    pub fn outputs(&self) -> &IOSet {
        &self.outputs
    }

    /// Set the initial values on the IOs in an `IOSet` using a set of Input Initializers
    fn set_initializers(
        &mut self,
        initializer_map: &BTreeMap<String, InputInitializer>,
    ) -> Result<()> {
        for (input_name, initializer) in initializer_map {
            // Go through all inputs matching names with names used in initializers
            for (index, input) in self.inputs.iter_mut().enumerate() {
                if input.name() == input_name || (input_name.as_str() == "default" && index == 0) {
                    input
                        .set_initializer(Some(initializer.clone()))
                        .chain_err(|| "Failed to set initializers in flow")?;
                }
            }
        }
        Ok(())
    }

    /// Configure a flow with additional information after it is deserialized from file
    ///
    /// # Errors
    ///
    /// Will return `Err` if the provided `initializations` cannot be set as initializers on
    /// the flow
    pub fn config(
        &mut self,
        source_url: &Url,
        parent_route: &Route,
        alias_from_reference: &Name,
        id: usize,
        initializations: &BTreeMap<String, InputInitializer>,
    ) -> Result<()> {
        self.id = id;
        self.set_alias(alias_from_reference);
        source_url.clone_into(&mut self.source_url);
        self.set_initializers(initializations)?;
        self.set_routes_from_parent(parent_route);
        self.validate()
    }

    /// Check if the flow can be run (it could be a sub-flow not a context level runnable flow)
    #[must_use]
    pub fn is_runnable(&self) -> bool {
        self.inputs().is_empty() && self.outputs().is_empty()
    }

    /// Find a `Process` in the flow hierarchy by its `Route`.
    ///
    /// The route is matched against this flow's own route first. If they match,
    /// returns `None` (the caller already has the flow). Otherwise, walks the
    /// `subprocesses` tree by stripping route segments to find the target process.
    #[must_use]
    pub fn process_from_route(&self, route: &Route) -> Option<&Process> {
        let sub = route.sub_route_of(&self.route)?;
        if sub.is_empty() {
            return None;
        }

        let segments: Vec<&str> = sub.as_ref().split('/').collect();
        let alias = *segments.first()?;
        let process = self.subprocesses.get(alias)?;

        let rest = segments.get(1..).unwrap_or_default();
        if rest.is_empty() {
            return Some(process);
        }

        match process {
            Process::FlowProcess(sub_flow) => {
                let remaining = Route::from(rest.join("/"));
                let mut child_route = self.route.clone();
                child_route.extend(&Route::from(alias));
                child_route.extend(&remaining);
                sub_flow.process_from_route(&child_route)
            }
            Process::FunctionProcess(_) => None,
        }
    }

    /// Mutable version of [`process_from_route`](Self::process_from_route).
    pub fn process_from_route_mut(&mut self, route: &Route) -> Option<&mut Process> {
        let sub = route.sub_route_of(&self.route)?;
        if sub.is_empty() {
            return None;
        }

        let segments: Vec<&str> = sub.as_ref().split('/').collect();
        let alias = *segments.first()?;

        let Some(rest) = segments.get(1..) else {
            return self.subprocesses.get_mut(alias);
        };
        if rest.is_empty() {
            return self.subprocesses.get_mut(alias);
        }

        let process = self.subprocesses.get_mut(alias)?;
        if let Process::FlowProcess(ref mut sub_flow) = process {
            let remaining = Route::from(rest.join("/"));
            let mut child_route = self.route.clone();
            child_route.extend(&Route::from(alias));
            child_route.extend(&remaining);
            sub_flow.process_from_route_mut(&child_route)
        } else {
            Some(process)
        }
    }

    fn get_subprocess_io(
        &mut self,
        subprocess_alias: &Name,
        direction: &Direction,
        sub_route: &Route,
    ) -> Result<IO> {
        debug!("\tLooking for subprocess with alias = '{subprocess_alias}'");

        // TODO create a trait HasInputs and HasOutputs and implement it for function and flow
        // and process so this below can avoid the match
        match self.subprocesses.get_mut(subprocess_alias) {
            Some(FlowProcess(ref mut sub_flow)) => {
                debug!("\tFlow sub-process with matching name found, name = '{subprocess_alias}'");
                match direction {
                    TO => sub_flow.inputs.find_by_subroute(sub_route),
                    FROM => sub_flow.outputs.find_by_subroute(sub_route),
                }
            }

            Some(FunctionProcess(ref mut function)) => {
                debug!("\tFunction sub-process with name = '{subprocess_alias}' found");
                match direction {
                    TO => function.inputs.find_by_subroute(sub_route),
                    FROM => function.outputs.find_by_subroute(sub_route).or_else(|e1| {
                        // for connections FROM the Input value copied at the output
                        function.inputs.find_by_subroute(sub_route).chain_err(|| e1)
                    }),
                }
            }

            None => {
                bail!(
                    "No sub-process named '{subprocess_alias}' exists in the flow '{}'\n\
                possible sub-process names are: '{}'",
                    self.route,
                    self.subprocesses
                        .keys()
                        .map(std::string::String::as_str)
                        .collect::<Vec<&str>>()
                        .join(", ")
                )
            }
        }
    }

    // Find an IO in a flow using its Route. Depending on the direction, possible IOs inside a flow are:
    //
    // FROM an Input of the Flow itself (FlowInput)
    // TO an Output of the Flow itself (FlowOutput)
    //
    // TO or FROM an Input/Output of a subprocess (subflow, or function) (SubProcess)
    //
    // Invalid Connections are:
    // FROM a FLowOutput
    // TO a FlowInput
    fn get_io_by_route(&mut self, direction: &Direction, route: &Route) -> Result<IO> {
        debug!("Looking for connection {direction:?} '{route}'");
        match (&direction, route.parse_subroute()?) {
            (&FROM, RouteType::FlowInput(input_name, sub_route)) => {
                let mut from = self
                    .inputs
                    .find_by_subroute(&Route::from(input_name.clone()))?;
                from.route_mut().extend(&sub_route);
                if !sub_route.is_empty() {
                    from.narrow_initializer(&sub_route);
                }
                Ok(from)
            }

            (&TO, RouteType::FlowOutput(output_name)) => self
                .outputs
                .find_by_subroute(&Route::from(output_name.clone())),

            (_, RouteType::SubProcess(process_name, sub_route)) => {
                self.get_subprocess_io(&process_name, direction, &sub_route)
            }

            (&FROM, RouteType::FlowOutput(output_name)) => {
                bail!("Invalid connection FROM an output named: '{}'", output_name)
            }

            (&TO, RouteType::FlowInput(input_name, sub_route)) => {
                bail!(
                    "Invalid connection TO an input named: '{}' with sub_route: '{}'",
                    input_name,
                    sub_route
                )
            }
        }
    }

    /// Iterate over all the connections defined in the flow, and attempt to connect the source
    /// and destination (within the flow), checking the two types are compatible
    ///
    /// # Errors
    ///
    /// Will return `Err`if some connections described within the flow cannot be build
    pub fn build_connections(&mut self, level: usize) -> Result<()> {
        debug!("Building connections for flow '{}'", self.name);

        let mut error_count = 0;

        // get connections out of self - so we can use immutable references to self inside loop
        let mut connections = take(&mut self.connections);

        for connection in &mut connections {
            if let Err(e) = self.build_connection(connection, level) {
                error_count += 1;
                error!("{e}");
            }
        }

        if error_count == 0 {
            debug!(
                "All connections inside flow '{}' successfully built",
                self.source_url
            );
            Ok(())
        } else {
            bail!(
                "{} connection errors found in flow '{}'",
                error_count,
                self.source_url
            )
        }
    }

    // Connection to/from Formats:
    // "input/input_name"       - An Input of this Flow
    // "output/output_name"     - An Output of this Flow
    // "flow_name/io_name"      - An Input or Output of a sub-flow within this flow
    // "function_name/io_name"  - An Input or an Output of a function within this flow
    //
    // Propagate any initializers on a flow output to the input (subflow or function) it is connected to
    fn build_connection(&mut self, connection: &Connection, level: usize) -> Result<()> {
        let from_io = self
            .get_io_by_route(&FROM, connection.from())
            .chain_err(|| {
                format!(
                    "Did not find connection source: '{}' specified in flow '{}'\n",
                    connection.from(),
                    self.source_url
                )
            })?;
        trace!("Found connection source:\n{from_io:#?}");

        // Connection can specify multiple destinations within flow - iterate over them all
        for to_route in connection.to() {
            match self.get_io_by_route(&TO, to_route) {
                Ok(to_io) => {
                    trace!("Found connection destination:\n{to_io:#?}");
                    let mut new_connection = connection.clone();
                    new_connection.connect(from_io.clone(), to_io, level)?;
                    self.connections.push(new_connection);
                }
                Err(error) => {
                    bail!(
                        "Did not find connection destination: '{}' in flow '{}'\n\t\t{}",
                        to_route,
                        self.source_url,
                        error
                    );
                }
            }
        }
        Ok(())
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod test {
    use std::collections::BTreeMap;

    use serde_json::json;

    use crate::model::connection::{Connection, Direction};
    use crate::model::datatype::{NUMBER_TYPE, STRING_TYPE};
    use crate::model::flow_definition::FlowDefinition;
    use crate::model::function_definition::FunctionDefinition;
    use crate::model::input::InputInitializer::Always;
    use crate::model::input::InputInitializer::Once;
    use crate::model::io::IO;
    use crate::model::name::{HasName, Name};
    use crate::model::process::Process;
    use crate::model::route::{HasRoute, Route, SetRoute};
    use crate::model::validation::Validate;

    // Create a test flow we can use in connection building testing
    fn test_flow() -> FlowDefinition {
        let mut flow = FlowDefinition {
            name: "test_flow".into(),
            alias: "test_flow".into(),
            inputs: vec![
                IO::new_named(vec![STRING_TYPE.into()], "string", "string"),
                IO::new_named(vec![NUMBER_TYPE.into()], "number", "number"),
            ],
            outputs: vec![
                IO::new_named(vec![STRING_TYPE.into()], "string", "string"),
                IO::new_named(vec![NUMBER_TYPE.into()], "number", "number"),
            ],
            source_url: FlowDefinition::default_url(),
            ..Default::default()
        };

        let process_1 = Process::FunctionProcess(FunctionDefinition {
            name: "process_1".into(),
            inputs: vec![IO::new_named(vec![STRING_TYPE.into()], "", "")],
            outputs: vec![IO::new_named(
                vec![STRING_TYPE.into()],
                "output_1",
                "output_1",
            )],
            ..Default::default()
        });

        let process_2 = Process::FunctionProcess(FunctionDefinition {
            name: "process_2".into(),
            process_id: 1,
            inputs: vec![IO::new_named(vec![STRING_TYPE.into()], "", "")],
            outputs: vec![IO::new_named(vec![NUMBER_TYPE.into()], "", "")],
            ..Default::default()
        });

        let _ = flow.subprocesses.insert("process_1".into(), process_1);
        let _ = flow.subprocesses.insert("process_2".into(), process_2);

        flow
    }

    #[test]
    fn test_name() {
        let flow = FlowDefinition::default();
        assert_eq!(flow.name(), &Name::default());
    }

    #[test]
    fn test_alias() {
        let flow = FlowDefinition::default();
        assert_eq!(flow.alias(), &Name::default());
    }

    #[test]
    fn test_set_alias() {
        let mut flow = FlowDefinition::default();
        flow.set_alias(&Name::from("test flow"));
        assert_eq!(flow.alias(), &Name::from("test flow"));
    }

    #[test]
    fn test_set_empty_alias() {
        let mut flow = FlowDefinition::default();
        flow.set_alias(&Name::from(""));
        assert_eq!(flow.alias(), &Name::from(""));
    }

    #[test]
    fn test_route() {
        let flow = FlowDefinition::default();
        assert_eq!(flow.route(), &Route::default());
    }

    #[test]
    fn test_non_existent_subprocess_in_connection() {
        let mut flow = test_flow();
        match flow.get_subprocess_io(
            &Name::from("foo"),
            &Direction::FROM,
            &Route::from("who-cares"),
        ) {
            Ok(_) => panic!("Should not find non-existent sub-process"),
            Err(e) => {
                assert!(e.to_string().contains("No sub-process named"));
            }
        }
    }

    #[test]
    fn test_existent_subprocess_existing_io_in_connection() {
        let mut flow = test_flow();
        flow.get_subprocess_io(&Name::from("process_1"), &Direction::FROM, &Route::from(""))
            .expect("Could not find sub-process called process_1");
    }

    #[test]
    fn test_existent_subprocess_non_existing_input_in_connection() {
        let mut flow = test_flow();
        match flow.get_subprocess_io(
            &Name::from("process_1"),
            &Direction::TO,
            &Route::from("no-such-io"),
        ) {
            Ok(_) => panic!("Should not find non-existent sub-process input"),
            Err(e) => assert!(e.to_string().contains("No IO")),
        }
    }

    #[test]
    fn test_existent_subprocess_non_existing_io_in_connection() {
        let mut flow = test_flow();
        match flow.get_subprocess_io(
            &Name::from("process_1"),
            &Direction::FROM,
            &Route::from("no-such-io"),
        ) {
            Ok(_) => panic!("Should not find non-existent sub-process IO"),
            Err(e) => {
                assert!(e.to_string().contains("No IO"));
            }
        }
    }

    #[test]
    fn test_route_mut() {
        let mut flow = FlowDefinition::default();
        let route = flow.route_mut();
        assert_eq!(route, &Route::default());
        *route = Route::from("/root");
        assert_eq!(route, &Route::from("/root"));
    }

    #[test]
    fn test_set_empty_parent_route() {
        let mut flow = test_flow();
        flow.set_routes_from_parent(&Route::from(""));
        assert_eq!(flow.route(), &Route::from("/test_flow"));
    }

    #[test]
    fn test_set_parent_route() {
        let mut flow = test_flow();
        flow.set_routes_from_parent(&Route::from("/root"));
        assert_eq!(flow.route(), &Route::from("/root/test_flow"));
    }

    #[test]
    fn validate_flow() {
        let mut flow = test_flow();
        let connection = Connection::new("process_1", "process_2");
        flow.connections = vec![connection];
        assert!(flow.validate().is_ok());
    }

    #[test]
    fn duplicate_connection() {
        let mut flow = test_flow();
        let connection = Connection::new("process_1", "process_2");
        flow.connections = vec![connection.clone(), connection];
        assert!(flow.validate().is_ok());
    }

    #[test]
    fn check_outputs() {
        let flow = test_flow();
        assert_eq!(flow.outputs().len(), 2);
    }

    #[test]
    fn check_inputs() {
        let flow = test_flow();
        assert_eq!(flow.inputs().len(), 2);
    }

    #[test]
    fn check_inputs_mut() {
        let mut flow = test_flow();
        let inputs = flow.inputs_mut();
        assert_eq!(inputs.len(), 2);
        *inputs = vec![];
        assert_eq!(inputs.len(), 0);
    }

    #[test]
    fn test_inputs_initializers() {
        let mut flow = test_flow();
        let mut initializers = BTreeMap::new();
        initializers.insert(STRING_TYPE.into(), Always(json!("Hello")));
        initializers.insert(NUMBER_TYPE.into(), Once(json!(42)));
        flow.set_initializers(&initializers)
            .expect("Could not set initializers");

        assert_eq!(
            flow.inputs()
                .first()
                .expect("Could not get input")
                .get_initializer()
                .as_ref()
                .expect("Could not get initializer"),
            &Always(json!("Hello"))
        );

        assert_eq!(
            flow.inputs()
                .get(1)
                .expect("Could not get input")
                .get_initializer()
                .as_ref()
                .expect("Could not get initializer"),
            &Once(json!(42))
        );
    }

    #[test]
    fn display_flow() {
        let mut flow = test_flow();
        let connection = Connection::new("process_1", "process_2");
        flow.connections = vec![connection];
        println!("flow: {flow}");
    }

    mod build_connection_tests {
        use crate::model::connection::Connection;
        use crate::model::flow_definition::test::test_flow;

        #[test]
        fn build_compatible_internal_connection() {
            let mut flow = test_flow();
            let connection = Connection::new("process_1", "process_2");
            assert!(flow.build_connection(&connection, 0).is_ok());
        }

        #[test]
        fn build_incompatible_internal_connection() {
            let mut flow = test_flow();
            let connection = Connection::new("process_2", "process_1");
            assert!(flow.build_connection(&connection, 0).is_err());
        }

        #[test]
        fn build_from_flow_input_to_sub_process() {
            let mut flow = test_flow();
            let connection = Connection::new("input/string", "process_1");
            assert!(flow.build_connection(&connection, 1).is_ok());
        }

        #[test]
        fn build_from_sub_process_flow_output() {
            let mut flow = test_flow();
            let connection = Connection::new("process_1", "output/string");
            assert!(flow.build_connection(&connection, 0).is_ok());
        }

        #[test]
        fn build_from_flow_input_to_flow_output() {
            let mut flow = test_flow();
            let connection = Connection::new("input/string", "output/string");
            assert!(flow.build_connection(&connection, 1).is_ok());
        }

        #[test]
        fn build_incompatible_from_flow_input_to_sub_process() {
            let mut flow = test_flow();
            let connection = Connection::new("input/number", "process_1");
            assert!(flow.build_connection(&connection, 1).is_err());
        }

        #[test]
        fn build_incompatible_from_sub_process_flow_output() {
            let mut flow = test_flow();
            let connection = Connection::new("process_1", "output/number");
            assert!(flow.build_connection(&connection, 0).is_err());
        }

        #[test]
        fn build_incompatible_from_flow_input_to_flow_output() {
            let mut flow = test_flow();
            let connection = Connection::new("input/string", "output/number");
            assert!(flow.build_connection(&connection, 1).is_err());
        }

        #[test]
        fn fail_build_from_flow_input_to_flow_input() {
            let mut flow = test_flow();
            let connection = Connection::new("input/string", "input/number");
            assert!(flow.build_connection(&connection, 1).is_err());
        }

        #[test]
        fn fail_build_from_flow_output_to_flow_output() {
            let mut flow = test_flow();
            let connection = Connection::new("output/string", "output/number");
            assert!(flow.build_connection(&connection, 1).is_err());
        }

        #[test]
        fn build_all_flow_connections() {
            let mut flow = test_flow();

            let connection1 = Connection::new("input/string", "output/string");
            let connection2 = Connection::new("input/string", "process_1");
            let connection3 = Connection::new("process_1", "output/string");

            flow.connections = vec![connection1, connection2, connection3];
            assert!(flow.build_connections(0).is_ok());
        }

        #[test]
        fn fail_build_flow_connections() {
            let mut flow = test_flow();
            let connection1 = Connection::new("input/number", "process_1");
            flow.connections = vec![connection1];
            assert!(flow.build_connections(0).is_err());
        }
    }

    #[test]
    fn deserialize_with_description() {
        use crate::deserializers::deserializer::get;
        use url::Url;

        let toml_str = r#"
        flow = "described_flow"
        description = "A flow that does something useful"
        "#;

        let url = Url::parse("file:///fake.toml").expect("Could not parse URL");
        let deserializer = get::<FlowDefinition>(&url).expect("Could not get deserializer");
        let flow: FlowDefinition = deserializer
            .deserialize(toml_str, Some(&url))
            .expect("Could not deserialize FlowDefinition with description");
        assert_eq!(flow.description, "A flow that does something useful");
    }

    #[test]
    fn deserialize_without_description() {
        use crate::deserializers::deserializer::get;
        use url::Url;

        let toml_str = r#"
        flow = "no_desc_flow"
        "#;

        let url = Url::parse("file:///fake.toml").expect("Could not parse URL");
        let deserializer = get::<FlowDefinition>(&url).expect("Could not get deserializer");
        let flow: FlowDefinition = deserializer
            .deserialize(toml_str, Some(&url))
            .expect("Could not deserialize FlowDefinition without description");
        assert_eq!(flow.description, "");
    }

    fn nested_test_flow() -> FlowDefinition {
        let mut inner_func = FunctionDefinition {
            name: "inner_func".into(),
            ..Default::default()
        };
        inner_func.route = Route::from("/root/sub_flow/inner_func");

        let mut sub_flow = FlowDefinition {
            name: "sub_flow".into(),
            alias: "sub_flow".into(),
            route: Route::from("/root/sub_flow"),
            ..Default::default()
        };
        sub_flow
            .subprocesses
            .insert("inner_func".into(), Process::FunctionProcess(inner_func));

        let mut func = FunctionDefinition {
            name: "top_func".into(),
            ..Default::default()
        };
        func.route = Route::from("/root/top_func");

        let mut root = FlowDefinition {
            name: "root".into(),
            alias: "root".into(),
            route: Route::from("/root"),
            ..Default::default()
        };
        root.subprocesses
            .insert("sub_flow".into(), Process::FlowProcess(sub_flow));
        root.subprocesses
            .insert("top_func".into(), Process::FunctionProcess(func));
        root
    }

    #[test]
    fn process_from_route_finds_top_level_function() {
        let root = nested_test_flow();
        let process = root.process_from_route(&Route::from("/root/top_func"));
        assert!(process.is_some());
        assert!(matches!(
            process.expect("process not found"),
            Process::FunctionProcess(_)
        ));
    }

    #[test]
    fn process_from_route_finds_sub_flow() {
        let root = nested_test_flow();
        let process = root.process_from_route(&Route::from("/root/sub_flow"));
        assert!(process.is_some());
        assert!(matches!(
            process.expect("process not found"),
            Process::FlowProcess(_)
        ));
    }

    #[test]
    fn process_from_route_finds_nested_function() {
        let root = nested_test_flow();
        let process = root.process_from_route(&Route::from("/root/sub_flow/inner_func"));
        assert!(process.is_some());
        assert!(matches!(
            process.expect("process not found"),
            Process::FunctionProcess(_)
        ));
    }

    #[test]
    fn process_from_route_returns_none_for_root() {
        let root = nested_test_flow();
        let process = root.process_from_route(&Route::from("/root"));
        assert!(process.is_none());
    }

    #[test]
    fn process_from_route_returns_none_for_nonexistent() {
        let root = nested_test_flow();
        let process = root.process_from_route(&Route::from("/root/nonexistent"));
        assert!(process.is_none());
    }

    #[test]
    fn process_from_route_returns_none_for_unrelated_route() {
        let root = nested_test_flow();
        let process = root.process_from_route(&Route::from("/other/path"));
        assert!(process.is_none());
    }

    #[test]
    fn process_from_route_mut_modifies_function() {
        let mut root = nested_test_flow();
        if let Some(Process::FunctionProcess(ref mut f)) =
            root.process_from_route_mut(&Route::from("/root/top_func"))
        {
            f.name = "renamed".into();
        }
        if let Some(Process::FunctionProcess(f)) =
            root.process_from_route(&Route::from("/root/top_func"))
        {
            assert_eq!(f.name, "renamed");
        } else {
            panic!("Expected to find renamed function");
        }
    }

    #[test]
    fn process_from_route_mut_modifies_nested() {
        let mut root = nested_test_flow();
        if let Some(Process::FunctionProcess(ref mut f)) =
            root.process_from_route_mut(&Route::from("/root/sub_flow/inner_func"))
        {
            f.name = "deep_rename".into();
        }
        if let Some(Process::FunctionProcess(f)) =
            root.process_from_route(&Route::from("/root/sub_flow/inner_func"))
        {
            assert_eq!(f.name, "deep_rename");
        } else {
            panic!("Expected to find renamed nested function");
        }
    }
}