Splitting test into unit and integration tests to match with Rust testing standards

.clippy.toml

@@ -1,3 +1,3 @@
-excessive-nesting-threshold = 4
+excessive-nesting-threshold = 5
 too-many-arguments-threshold = 10
 allowed-idents-below-min-chars = ["..", "k", "v", "f", "re", "id", "Ok", "'_"]

.github/workflows/tests.yaml

@@ -17,7 +17,7 @@ jobs:
       - name: Install Rust + components
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-          toolchain: 1.87
+          toolchain: 1.89
           components: rustfmt,clippy
       - name: Install code coverage
         uses: taiki-e/install-action@cargo-llvm-cov

.vscode/settings.json

@@ -9,13 +9,8 @@
     ],
     "files.autoSave": "off",
     "files.insertFinalNewline": true,
-    "gitlens.showWelcomeOnInstall": false,
     "gitlens.showWhatsNewAfterUpgrades": false,
     "lldb.consoleMode": "evaluate",
-    "rust-analyzer.cargo.features": [
-        "test"
-    ],
-    "rust-analyzer.cargo.noDefaultFeatures": true,
     "rust-analyzer.check.command": "clippy",
     "rust-analyzer.runnables.extraTestBinaryArgs": [
         "--nocapture"

Cargo.toml

@@ -55,6 +55,7 @@ glob = "0.3.1"
 heck = "0.5.0"
 # convert bytes to hex strings
 hex = "0.4.3"
+hostname = "0.4.1"
 # hashmaps that preserve insertion order
 indexmap = { version = "2.9.0", features = ["serde"] }
 # utilities for iterables e.g. cartesian products

cspell.json

@@ -80,6 +80,9 @@
         "wasi",
         "patchelf",
         "itertools",
+        "colinianking",
+        "itertools",
+        "pathset",
         "colinianking"
     ],
     "useGitignore": false,

notebook.ipynb

@@ -0,0 +1,65 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "id": "3ba2aae6",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import orcapod as op"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "4d867643",
+   "metadata": {},
+   "outputs": [
+    {
+     "ename": "TypeError",
+     "evalue": "Pipeline.__init__() missing 4 required positional arguments: 'graph_dot', 'metadata', 'input_spec', and 'output_spec'",
+     "output_type": "error",
+     "traceback": [
+      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
+      "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
+      "Cell \u001b[0;32mIn[3], line 1\u001b[0m\n\u001b[0;32m----> 1\u001b[0m \u001b[43mop\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mPipeline\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
+      "\u001b[0;31mTypeError\u001b[0m: Pipeline.__init__() missing 4 required positional arguments: 'graph_dot', 'metadata', 'input_spec', and 'output_spec'"
+     ]
+    }
+   ],
+   "source": [
+    "op.Pipeline()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "7626744f",
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "orcapod",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.10.18"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

src/core/crypto.rs

@@ -107,3 +107,43 @@ pub fn make_random_hash() -> String {
     rand::rng().fill_bytes(&mut bytes);
     hex::encode(bytes)
 }
+
+#[cfg(test)]
+mod tests {
+    #![expect(clippy::panic_in_result_fn, reason = "OK in tests.")]
+    use crate::{
+        core::crypto::{hash_buffer, hash_dir, hash_file},
+        uniffi::error::Result,
+    };
+    use std::fs::read;
+
+    #[test]
+    fn consistent_hash() -> Result<()> {
+        let filepath = "./tests/extra/data/images/subject.jpeg";
+        assert_eq!(
+            hash_file(filepath)?,
+            hash_buffer(&read(filepath)?),
+            "Checksum not consistent."
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn complex_hash() -> Result<()> {
+        let dirpath = "./tests/extra/data/images";
+        assert_eq!(
+            hash_dir(dirpath)?,
+            "6c96a478ea25e34fab045bc82858a2980b2cfb22db32e83c01349a8e7ed3b42c".to_owned(),
+            "Directory checksum didn't match."
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn internal_invalid_filepath() {
+        assert!(
+            hash_file("nonexistent_file.txt").is_err(),
+            "Did not raise an invalid filepath error."
+        );
+    }
+}

src/core/error.rs

@@ -123,7 +123,13 @@ impl fmt::Debug for OrcaError {
             Kind::AgentCommunicationFailure { backtrace, .. }
             | Kind::IncompletePacket { backtrace, .. }
             | Kind::InvalidFilepath { backtrace, .. }
+            | Kind::InvalidIndex { backtrace, .. }
+            | Kind::KeyMissing { backtrace, .. }
             | Kind::MissingInfo { backtrace, .. }
+            | Kind::FailedToGetPodJobOutput { backtrace, .. }
+            | Kind::PodJobProcessingError { backtrace, .. }
+            | Kind::PodJobSubmissionFailed { backtrace, .. }
+            | Kind::UnexpectedPathType { backtrace, .. }
             | Kind::BollardError { backtrace, .. }
             | Kind::ChronoParseError { backtrace, .. }
             | Kind::DOTError { backtrace, .. }

src/core/graph.rs

@@ -1,5 +1,5 @@
 use crate::{
-    core::{pipeline::PipelineNode, util::get},
+    core::{model::pipeline::PipelineNode, util::get},
     uniffi::{error::Result, model::pipeline::Kernel},
 };
 use dot_parser::ast::Graph as DOTGraph;
@@ -25,7 +25,7 @@ pub fn make_graph(
     let graph =
         DiGraph::<DotNodeWeight, DotAttrList>::from_dot_graph(DOTGraph::try_from(input_dot)?).map(
             |_, node| PipelineNode {
-                name: node.id.clone(),
+                id: node.id.clone(),
                 kernel: get(&metadata, &node.id)
                     .unwrap_or_else(|error| panic!("{error}"))
                     .clone(),

src/core/mod.rs

@@ -1,40 +1,14 @@
-macro_rules! inner_attr_to_each {
-    { #!$attr:tt $($it:item)* } => {
-        $(
-            #$attr
-            $it
-        )*
-    }
-}
-
 pub(crate) mod error;
 pub(crate) mod graph;
-pub(crate) mod pipeline;
 pub(crate) mod store;
 pub(crate) mod util;
 pub(crate) mod validation;
 
-inner_attr_to_each! {
-    #![cfg(feature = "default")]
-    pub(crate) mod crypto;
-    pub(crate) mod model;
-    pub(crate) mod operator;
-    pub(crate) mod orchestrator;
-}
+pub(crate) mod crypto;
+/// Model definition for orcapod
+pub mod model;
+pub(crate) mod operator;
+pub(crate) mod orchestrator;
 
-#[cfg(feature = "test")]
-inner_attr_to_each! {
-    #![cfg_attr(
-        feature = "test",
-        allow(
-            missing_docs,
-            clippy::missing_errors_doc,
-            clippy::missing_panics_doc,
-            reason = "Documentation not necessary since private API.",
-        ),
-    )]
-    pub mod crypto;
-    pub mod model;
-    pub mod operator;
-    pub mod orchestrator;
-}
+/// Pipeline runner module
+pub mod pipeline_runner;

src/core/model/mod.rs

@@ -32,7 +32,7 @@ pub fn to_yaml<T: Serialize>(instance: &T) -> Result<String> {
     Ok(yaml)
 }
 
-pub fn serialize_hashmap<S, K: Ord + Serialize, V: Serialize, BH: BuildHasher>(
+pub(crate) fn serialize_hashmap<S, K: Ord + Serialize, V: Serialize, BH: BuildHasher>(
     map: &HashMap<K, V, BH>,
     serializer: S,
 ) -> result::Result<S::Ok, S::Error>
@@ -44,7 +44,7 @@ where
 }
 
 #[allow(clippy::ref_option, reason = "Serde requires this signature.")]
-pub fn serialize_hashmap_option<S, K: Ord + Serialize, V: Serialize, BH: BuildHasher>(
+pub(crate) fn serialize_hashmap_option<S, K: Ord + Serialize, V: Serialize, BH: BuildHasher>(
     map_option: &Option<HashMap<K, V, BH>>,
     serializer: S,
 ) -> result::Result<S::Ok, S::Error>
@@ -57,4 +57,5 @@ where
     sorted.serialize(serializer)
 }
 
-pub mod pod;
+pub(crate) mod pipeline;
+pub(crate) mod pod;

src/core/model/pipeline.rs

@@ -0,0 +1,109 @@
+use std::{
+    backtrace::Backtrace,
+    collections::{HashMap, HashSet},
+};
+
+use crate::uniffi::{
+    error::{Kind, OrcaError, Result},
+    model::{
+        packet::PathSet,
+        pipeline::{Kernel, Pipeline, PipelineJob},
+    },
+};
+use itertools::Itertools as _;
+use petgraph::Direction::Incoming;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
+pub struct PipelineNode {
+    pub id: String,
+    pub kernel: Kernel,
+}
+
+impl Pipeline {
+    /// Function to get the parents of a node
+    pub(crate) fn get_node_parents(
+        &self,
+        node: &PipelineNode,
+    ) -> impl Iterator<Item = &PipelineNode> {
+        // Find the NodeIndex for the given node_key
+        let node_index = self
+            .graph
+            .node_indices()
+            .find(|&idx| self.graph[idx] == *node);
+        node_index.into_iter().flat_map(move |idx| {
+            self.graph
+                .neighbors_directed(idx, Incoming)
+                .map(move |parent_idx| &self.graph[parent_idx])
+        })
+    }
+
+    /// Return a vec of `node_names` that takes in inputs based on the `input_spec`
+    pub(crate) fn get_input_nodes(&self) -> HashSet<&String> {
+        let mut input_nodes = HashSet::new();
+
+        self.input_spec.iter().for_each(|(_, node_uris)| {
+            for node_uri in node_uris {
+                input_nodes.insert(&node_uri.node_id);
+            }
+        });
+
+        input_nodes
+    }
+}
+
+impl PipelineJob {
+    /// Helpful function to get the input packet for input nodes of the pipeline based on the `pipeline_job` an`pipeline_spec`ec
+    /// # Errors
+    /// Will return `Err` if there is an issue getting the input packet per node.
+    /// # Returns
+    /// A `HashMap` where the key is the node name and the value is a vector of `HashMap<String, PathSet>` representing the input packets for that node.
+    pub fn get_input_packet_per_node(
+        &self,
+    ) -> Result<HashMap<String, Vec<HashMap<String, PathSet>>>> {
+        // For each node in the input specification, we will iterate over its mapping
+        // nodes_input_spec contains <node_id, HashMap<key, PathSet>>
+        let mut nodes_input_spec = HashMap::new();
+        for (input_key, node_uris) in &self.pipeline.input_spec {
+            for node_uri in node_uris {
+                let input_path_sets = self.input_packet.get(input_key).ok_or(OrcaError {
+                    kind: Kind::KeyMissing {
+                        key: input_key.clone(),
+                        backtrace: Some(Backtrace::capture()),
+                    },
+                })?;
+                // There shouldn't be a duplicate key in the input packet as this will be handle by pipeline verify
+                let input_spec = nodes_input_spec
+                    .entry(&node_uri.node_id)
+                    .or_insert_with(HashMap::new);
+                input_spec.insert(&node_uri.key, input_path_sets);
+            }
+        }
+
+        // For each node, compute the cartesian product of the path_sets for each unique combination of keys
+        let node_input_packets = nodes_input_spec
+            .into_iter()
+            .map(|(node_id, input_node_keys)| {
+                // We need to pull them out at the same time to ensure the key order is preserve to match the cartesian product
+                let (keys, values): (Vec<_>, Vec<_>) = input_node_keys.into_iter().unzip();
+
+                // Covert each combo into a packet
+                let packets = values
+                    .into_iter()
+                    .multi_cartesian_product()
+                    .map(|combo| {
+                        keys.iter()
+                            .copied()
+                            .zip(combo)
+                            .map(|(key, pathset)| (key.to_owned(), pathset.to_owned()))
+                            .collect::<HashMap<_, _>>()
+                    })
+                    .collect::<Vec<HashMap<String, PathSet>>>();
+
+                (node_id.to_owned(), packets)
+            })
+            .collect::<HashMap<_, _>>();
+
+        Ok(node_input_packets)
+    }
+}