[dagster-pipes-rust] - Inital integration with the pipes test suite

.github/workflows/libraries-pipes-rust.yml

@@ -40,12 +40,18 @@ jobs:
         with:
           tool: nextest
 
+      - name: "Install uv"
+        uses: astral-sh/setup-uv@v4
+
       - name: "Tests"
         run: cargo nextest run
 
       - name: "Doctests"
         run: cargo test --doc
 
+      - name: "Python tests"
+        run: uv run pytest
+
   integration:
     name: "integration"
     runs-on: ubuntu-latest

libraries/pipes/implementations/rust/Cargo.toml

@@ -11,10 +11,18 @@ serde_json = "1"
 base64 = "0"
 flate2 = "1"
 thiserror = "2.0.3"
+clap = { version = "4.5.23", features = ["derive"], optional = true}
 
 [dev-dependencies]
 rstest = { version = "0.23.0", default-features = false }
 tempfile = "3.14.0"
 
 [workspace]
 members = ["example-dagster-pipes-rust-project/rust_processing_jobs"]
+
+[features]
+pipes-tests = ["dep:clap"]
+
+[[bin]]
+name = "pipes_tests"
+required-features = ["pipes-tests"]

libraries/pipes/implementations/rust/pipes.toml

@@ -0,0 +1,13 @@
+[general]
+error_reporting = false
+
+[messages]
+log = false
+report_custom_message = false
+report_asset_materialization = false
+report_asset_check = false
+log_external_stream = false
+
+[message_channel]
+s3 = false
+databricks = false

libraries/pipes/implementations/rust/pyproject.toml

@@ -0,0 +1,26 @@
+[project]
+name = "dagster-pipes-rust"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "dagster>=1.8.9",
+]
+
+[tool.uv]
+dev-dependencies = [
+    "dagster-webserver>=1.8.9",
+    "pyright>=1.1.383",
+    "pytest>=8.3.3",
+    "ruff>=0.6.8",
+    "dagster-pipes-tests",
+]
+
+[tool.uv.sources]
+dagster-pipes-tests = { path = "../../tests/dagster-pipes-tests", editable = true }
+
+[tool.pytest.ini_options]
+addopts = "-s "
+color = "yes"
+testpaths = ["tests"]

libraries/pipes/implementations/rust/src/bin/pipes_tests.rs

@@ -0,0 +1,88 @@
+use clap::ArgAction;
+use clap::Parser;
+use dagster_pipes_rust::{open_dagster_pipes, DagsterPipesError};
+use dagster_pipes_rust::{DAGSTER_PIPES_CONTEXT_ENV_VAR, DAGSTER_PIPES_MESSAGES_ENV_VAR};
+use std::collections::HashMap;
+use std::fs::File;
+
+#[derive(Parser)]
+struct Cli {
+    #[arg(long)]
+    context: Option<String>,
+    #[arg(long)]
+    messages: Option<String>,
+    #[arg(
+        long,
+        action = ArgAction::Set,
+        default_value_t = false,
+        default_missing_value = "false",
+        num_args=0..=1,
+        require_equals = false,
+    )]
+    env: bool,
+    #[arg(long = "jobName")]
+    job_name: Option<String>,
+    #[arg(long)]
+    extras: Option<String>,
+    #[arg(
+        long,
+        action = ArgAction::Set,
+        default_value_t = false,
+        default_missing_value = "false",
+        num_args=0..=1,
+        require_equals = false,
+    )]
+    full: bool,
+    #[arg(long)]
+    custom_payload_path: Option<String>,
+    #[arg(long)]
+    report_asset_check: Option<String>,
+    #[arg(long)]
+    report_asset_materialization: Option<String>,
+    #[arg(
+        long,
+        action = ArgAction::Set,
+        default_value_t = false,
+        default_missing_value = "false",
+        num_args=0..=1,
+        require_equals = false,
+    )]
+    throw_error: bool,
+    #[arg(
+        long,
+        action = ArgAction::Set,
+        default_value_t = false,
+        default_missing_value = "false",
+        num_args=0..=1,
+        require_equals = false,
+    )]
+    logging: bool,
+    #[arg(long)]
+    message_writer: Option<String>,
+    #[arg(long)]
+    context_loader: Option<String>,
+}
+
+pub fn main() -> Result<(), DagsterPipesError> {
+    let args = Cli::parse();
+    if let Some(context) = args.context {
+        std::env::set_var(DAGSTER_PIPES_CONTEXT_ENV_VAR, &context);
+    }
+    if let Some(messages) = args.messages {
+        std::env::set_var(DAGSTER_PIPES_MESSAGES_ENV_VAR, &messages);
+    }
+
+    let context = open_dagster_pipes()?;
+
+    if let Some(job_name) = args.job_name {
+        assert_eq!(context.data.job_name, Some(job_name));
+    }
+
+    if let Some(extras) = args.extras {
+        let file = File::open(extras).expect("extras could not be opened");
+        let json: HashMap<std::string::String, std::option::Option<serde_json::Value>> =
+            serde_json::from_reader(file).expect("extras could not be parsed");
+        assert_eq!(context.data.extras, Some(json));
+    }
+    Ok(())
+}

libraries/pipes/implementations/rust/src/lib.rs

@@ -20,7 +20,9 @@ use crate::writer::message_writer::{
 use crate::writer::message_writer_channel::MessageWriteError;
 
 pub use crate::context_loader::LoadContext;
-pub use crate::params_loader::LoadParams;
+pub use crate::params_loader::{
+    LoadParams, DAGSTER_PIPES_CONTEXT_ENV_VAR, DAGSTER_PIPES_MESSAGES_ENV_VAR,
+};
 pub use crate::types::{AssetCheckSeverity, PipesMetadataValue};
 pub use crate::writer::message_writer::{DefaultWriter, MessageWriter};
 pub use crate::writer::message_writer_channel::MessageWriterChannel;
@@ -34,7 +36,7 @@ pub struct PipesContext<W>
 where
     W: MessageWriter,
 {
-    data: PipesContextData,
+    pub data: PipesContextData,
     message_channel: W::Channel,
 }
 

libraries/pipes/implementations/rust/src/params_loader.rs

@@ -6,8 +6,8 @@ use std::fmt;
 use std::io::Read;
 use thiserror::Error;
 
-const DAGSTER_PIPES_CONTEXT_ENV_VAR: &str = "DAGSTER_PIPES_CONTEXT";
-const DAGSTER_PIPES_MESSAGES_ENV_VAR: &str = "DAGSTER_PIPES_MESSAGES";
+pub const DAGSTER_PIPES_CONTEXT_ENV_VAR: &str = "DAGSTER_PIPES_CONTEXT";
+pub const DAGSTER_PIPES_MESSAGES_ENV_VAR: &str = "DAGSTER_PIPES_MESSAGES";
 
 /// Load params passed from the orchestration process by the context injector and
 /// message reader. These params are used to respectively bootstrap implementations of

libraries/pipes/implementations/rust/tests/conftest.py

@@ -0,0 +1,7 @@
+import pytest
+import subprocess
+
+
+@pytest.fixture(scope="session", autouse=True)
+def built_binary():
+    subprocess.run(["cargo", "build", "--features", "pipes-tests"], check=True)

libraries/pipes/implementations/rust/tests/test_pipes.py

@@ -0,0 +1,6 @@
+from dagster_pipes_tests import PipesTestSuite
+
+
+class TestRustPipes(PipesTestSuite):
+    BASE_ARGS = ["./target/debug/pipes_tests"]
+