Pipeline Validation

.clippy.toml

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

.github/workflows/tests.yaml

@@ -17,12 +17,12 @@ jobs:
       - name: Install Rust + components
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-          toolchain: 1.87
+          toolchain: 1.90.0
           components: rustfmt,clippy
       - name: Install code coverage
         uses: taiki-e/install-action@cargo-llvm-cov
       - name: Run syntax and style tests
-        run: cargo clippy --no-default-features --features=test --all-targets -- -D warnings
+        run: cargo clippy --all-targets -- -D warnings
       - name: Run format test
         run: cargo fmt --check
       - name: Run integration tests w/ coverage report
@@ -57,15 +57,11 @@ jobs:
           uv pip install eclipse-zenoh -p ~/.local/share/base
           . ~/.local/share/base/bin/activate
           maturin develop --uv
-      - name: Run smoke test
+      - name: Run Python tests
         env:
           RUST_BACKTRACE: full
         run: |
           . ~/.local/share/base/bin/activate
-          python tests/extra/python/smoke_test.py -- tests/.tmp
-      - name: Run agent test
-        env:
-          RUST_BACKTRACE: full
-        run: |
-          . ~/.local/share/base/bin/activate
-          python tests/extra/python/agent_test.py -- tests/.tmp
+          for py_file in tests/extra/python/*.py; do
+          python "$py_file" -- tests/.tmp
+          done

.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
@@ -82,7 +83,7 @@ tokio = { version = "1.41.0", features = ["full"] }
 # utilities for async calls
 tokio-util = "0.7.13"
 # automated CFFI + bindings in other languages
-uniffi = { version = "0.29.1", features = ["cli", "tokio"] }
+uniffi = { version = "0.29.4", features = ["cli", "tokio"] }
 # shared, distributed memory via communication
 zenoh = "1.3.4"
 

cspell.json

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

src/core/crypto.rs

@@ -50,11 +50,9 @@ pub fn hash_buffer(buffer: impl AsRef<[u8]>) -> String {
 ///
 /// Will return error if unable to access file.
 pub fn hash_file(filepath: impl AsRef<Path>) -> Result<String> {
-    hash_stream(
-        &mut File::open(&filepath).context(selector::InvalidFilepath {
-            path: filepath.as_ref(),
-        })?,
-    )
+    hash_stream(&mut File::open(&filepath).context(selector::InvalidPath {
+        path: filepath.as_ref(),
+    })?)
 }
 /// Evaluate checksum hash of a directory.
 ///
@@ -107,3 +105,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

@@ -121,9 +121,20 @@ impl fmt::Debug for OrcaError {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match &self.kind {
             Kind::AgentCommunicationFailure { backtrace, .. }
+            | Kind::EmptyDir { backtrace, .. }
             | Kind::IncompletePacket { backtrace, .. }
-            | Kind::InvalidFilepath { backtrace, .. }
+            | Kind::InvalidPath { backtrace, .. }
+            | Kind::InvalidIndex { backtrace, .. }
+            | Kind::InvalidInputSpecNodeNotInGraph { backtrace, .. }
+            | Kind::InvalidOutputSpecKeyNotInNode { backtrace, .. }
+            | Kind::InvalidOutputSpecNodeNotInGraph { backtrace, .. }
+            | Kind::KeyMissing { backtrace, .. }
             | Kind::MissingInfo { backtrace, .. }
+            | Kind::FailedToGetPodJobOutput { backtrace, .. }
+            | Kind::PipelineValidationErrorMissingKeys { 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;
@@ -10,7 +10,6 @@ use petgraph::{
 use std::collections::HashMap;
 
 #[expect(
-    clippy::needless_pass_by_value,
     clippy::panic_in_result_fn,
     clippy::panic,
     reason = "
@@ -20,15 +19,17 @@ use std::collections::HashMap;
 )]
 pub fn make_graph(
     input_dot: &str,
-    metadata: HashMap<String, Kernel>,
+    metadata: &HashMap<String, Kernel>,
 ) -> Result<DiGraph<PipelineNode, ()>> {
     let graph =
         DiGraph::<DotNodeWeight, DotAttrList>::from_dot_graph(DOTGraph::try_from(input_dot)?).map(
-            |_, node| PipelineNode {
-                name: node.id.clone(),
-                kernel: get(&metadata, &node.id)
+            |node_idx, node| PipelineNode {
+                hash: String::new(),
+                kernel: get(metadata, &node.id)
                     .unwrap_or_else(|error| panic!("{error}"))
                     .clone(),
+                label: node.id.clone(),
+                node_idx,
             },
             |_, _| (),
         );

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(crate) 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

@@ -8,28 +8,33 @@ use std::{
     hash::BuildHasher,
     result,
 };
-/// Converts a model instance into a consistent yaml.
-///
-/// # Errors
-///
-/// Will return `Err` if there is an issue converting an `instance` into YAML (w/o annotation).
-pub fn to_yaml<T: Serialize>(instance: &T) -> Result<String> {
-    let mapping: IndexMap<String, Value> = serde_yaml::from_str(&serde_yaml::to_string(instance)?)?; // cast to map
-    let mut yaml = serde_yaml::to_string(
-        &mapping
-            .iter()
-            .filter_map(|(k, v)| match &**k {
-                "annotation" | "hash" => None,
-                "pod" | "pod_job" => Some((k, v["hash"].clone())),
-                _ => Some((k, v.clone())),
-            })
-            .collect::<IndexMap<_, _>>(),
-    )?; // skip fields and convert refs to hash pointers
-    yaml.insert_str(
-        0,
-        &format!("class: {}\n", get_type_name::<T>().to_snake_case()),
-    ); // replace class at top
-    Ok(yaml)
+
+/// Trait to handle serialization to yaml for `OrcaPod` models
+pub trait ToYaml: Serialize + Sized {
+    /// Serializes the instance to a YAML string.
+    /// # Errors
+    /// Will return `Err` if it fail to serialize instance to string
+    fn to_yaml(&self) -> Result<String> {
+        let mapping: IndexMap<String, Value> = serde_yaml::from_str(&serde_yaml::to_string(self)?)?; // cast to map
+        let mut yaml = serde_yaml::to_string(
+            &mapping
+                .iter()
+                .filter_map(|(k, v)| Self::process_field(k, v))
+                .collect::<IndexMap<_, _>>(),
+        )?; // skip fields and convert refs to hash pointers
+        yaml.insert_str(
+            0,
+            &format!("class: {}\n", get_type_name::<Self>().to_snake_case()),
+        ); // replace class at top
+        Ok(yaml)
+    }
+
+    /// Filter out which field to serialize and which to omit
+    ///
+    /// # Returns
+    /// (`field_name`, `field_value`): to be pass to `to_yaml` for serialization
+    /// None: to skip
+    fn process_field(field_name: &str, field_value: &Value) -> Option<(String, Value)>;
 }
 
 pub fn serialize_hashmap<S, K: Ord + Serialize, V: Serialize, BH: BuildHasher>(
@@ -57,4 +62,5 @@ where
     sorted.serialize(serializer)
 }
 
+pub mod pipeline;
 pub mod pod;