chore(derive): Pipeline Core Test Coverage (#642)

* feat(derive): pipeline core

* fix(derive): kona providers test utils

* fix(derive): test-utils

crates/derive/Cargo.toml

@@ -61,6 +61,7 @@ proptest.workspace = true
 tracing-subscriber.workspace = true
 alloy-node-bindings.workspace = true
 serde_json.workspace = true
+kona-providers = { workspace = true, features = ["test-utils"] }
 
 [features]
 default = ["serde"]
@@ -85,5 +86,6 @@ test-utils = [
   "dep:alloy-node-bindings",
   "dep:tracing-subscriber",
   "dep:alloy-rpc-client",
+  "kona-providers/test-utils",
   "alloy-transport-http/reqwest"
 ]

crates/derive/src/lib.rs

@@ -30,7 +30,10 @@ pub mod sources;
 pub mod stages;
 pub mod traits;
 
+mod macros;
+
 #[cfg(feature = "metrics")]
 pub mod metrics;
 
-mod macros;
+#[cfg(any(test, feature = "test-utils"))]
+pub mod test_utils;

crates/derive/src/pipeline/core.rs

@@ -195,3 +195,121 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test_utils::*;
+    use alloy_rpc_types_engine::PayloadAttributes;
+    use op_alloy_genesis::SystemConfig;
+    use op_alloy_rpc_types_engine::OptimismPayloadAttributes;
+
+    fn default_test_payload_attributes() -> OptimismAttributesWithParent {
+        OptimismAttributesWithParent {
+            attributes: OptimismPayloadAttributes {
+                payload_attributes: PayloadAttributes {
+                    timestamp: 0,
+                    prev_randao: Default::default(),
+                    suggested_fee_recipient: Default::default(),
+                    withdrawals: None,
+                    parent_beacon_block_root: None,
+                },
+                transactions: None,
+                no_tx_pool: None,
+                gas_limit: None,
+                eip_1559_params: None,
+            },
+            parent: Default::default(),
+            is_last_in_span: false,
+        }
+    }
+
+    #[test]
+    fn test_pipeline_next_attributes_empty() {
+        let mut pipeline = new_test_pipeline();
+        let result = pipeline.next();
+        assert_eq!(result, None);
+    }
+
+    #[test]
+    fn test_pipeline_next_attributes_with_peek() {
+        let mut pipeline = new_test_pipeline();
+        let expected = default_test_payload_attributes();
+        pipeline.prepared.push_back(expected.clone());
+
+        let result = pipeline.peek();
+        assert_eq!(result, Some(&expected));
+
+        let result = pipeline.next();
+        assert_eq!(result, Some(expected));
+    }
+
+    #[tokio::test]
+    async fn test_derivation_pipeline_missing_block() {
+        let mut pipeline = new_test_pipeline();
+        let cursor = L2BlockInfo::default();
+        let result = pipeline.step(cursor).await;
+        assert_eq!(
+            result,
+            StepResult::OriginAdvanceErr(
+                PipelineError::Provider("Block not found".to_string()).temp()
+            )
+        );
+    }
+
+    #[tokio::test]
+    async fn test_derivation_pipeline_prepared_attributes() {
+        let rollup_config = Arc::new(RollupConfig::default());
+        let l2_chain_provider = TestL2ChainProvider::default();
+        let expected = default_test_payload_attributes();
+        let attributes = TestNextAttributes { next_attributes: Some(expected) };
+        let mut pipeline = DerivationPipeline::new(attributes, rollup_config, l2_chain_provider);
+
+        // Step on the pipeline and expect the result.
+        let cursor = L2BlockInfo::default();
+        let result = pipeline.step(cursor).await;
+        assert_eq!(result, StepResult::PreparedAttributes);
+    }
+
+    #[tokio::test]
+    async fn test_derivation_pipeline_advance_origin() {
+        let rollup_config = Arc::new(RollupConfig::default());
+        let l2_chain_provider = TestL2ChainProvider::default();
+        let attributes = TestNextAttributes::default();
+        let mut pipeline = DerivationPipeline::new(attributes, rollup_config, l2_chain_provider);
+
+        // Step on the pipeline and expect the result.
+        let cursor = L2BlockInfo::default();
+        let result = pipeline.step(cursor).await;
+        assert_eq!(result, StepResult::AdvancedOrigin);
+    }
+
+    #[tokio::test]
+    async fn test_derivation_pipeline_signal_reset_missing_sys_config() {
+        let rollup_config = Arc::new(RollupConfig::default());
+        let l2_chain_provider = TestL2ChainProvider::default();
+        let attributes = TestNextAttributes::default();
+        let mut pipeline = DerivationPipeline::new(attributes, rollup_config, l2_chain_provider);
+
+        // Signal the pipeline to reset.
+        let l2_safe_head = L2BlockInfo::default();
+        let l1_origin = BlockInfo::default();
+        let result = pipeline.signal(Signal::Reset { l2_safe_head, l1_origin }).await.unwrap_err();
+        assert_eq!(result, PipelineError::Provider("System config not found".to_string()).temp());
+    }
+
+    #[tokio::test]
+    async fn test_derivation_pipeline_signal_reset_ok() {
+        let rollup_config = Arc::new(RollupConfig::default());
+        let mut l2_chain_provider = TestL2ChainProvider::default();
+        l2_chain_provider.system_configs.insert(0, SystemConfig::default());
+        let attributes = TestNextAttributes::default();
+        let mut pipeline = DerivationPipeline::new(attributes, rollup_config, l2_chain_provider);
+
+        // Signal the pipeline to reset.
+        let l2_safe_head = L2BlockInfo::default();
+        let l1_origin = BlockInfo::default();
+        let result = pipeline.signal(Signal::Reset { l2_safe_head, l1_origin }).await;
+        assert!(result.is_ok());
+    }
+}

crates/derive/src/test_utils/mod.rs

@@ -0,0 +1,114 @@
+//! Test Utilities for `kona-derive`.
+//!
+//! This includes top-level [crate::pipeline::DerivationPipeline]
+//! test utilities as well as individual stage test utilities.
+
+use alloc::{boxed::Box, sync::Arc};
+use op_alloy_genesis::{RollupConfig, SystemConfig};
+use op_alloy_protocol::{BlockInfo, L2BlockInfo};
+use op_alloy_rpc_types_engine::OptimismAttributesWithParent;
+
+// Re-export these types used internally to the test pipeline.
+pub use crate::{
+    batch::SingleBatch,
+    errors::PipelineError,
+    pipeline::{DerivationPipeline, PipelineBuilder, PipelineResult},
+    stages::{
+        test_utils::MockAttributesBuilder, AttributesProvider, AttributesQueue, BatchQueue,
+        BatchStream, ChannelBank, ChannelReader, FrameQueue, L1Retrieval, L1Traversal,
+    },
+    traits::{
+        test_utils::TestDAP, FlushableStage, NextAttributes, OriginAdvancer, OriginProvider,
+        ResettableStage,
+    },
+};
+pub use kona_providers::test_utils::{TestChainProvider, TestL2ChainProvider};
+
+/// A fully custom [NextAttributes].
+#[derive(Default, Debug, Clone)]
+pub struct TestNextAttributes {
+    /// The next [OptimismAttributesWithParent] to return.
+    pub next_attributes: Option<OptimismAttributesWithParent>,
+}
+
+#[async_trait::async_trait]
+impl FlushableStage for TestNextAttributes {
+    /// Flushes the stage.
+    async fn flush_channel(&mut self) -> PipelineResult<()> {
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl ResettableStage for TestNextAttributes {
+    /// Resets the derivation stage to its initial state.
+    async fn reset(&mut self, _: BlockInfo, _: &SystemConfig) -> PipelineResult<()> {
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl OriginProvider for TestNextAttributes {
+    /// Returns the current origin.
+    fn origin(&self) -> Option<BlockInfo> {
+        Some(BlockInfo::default())
+    }
+}
+
+#[async_trait::async_trait]
+impl OriginAdvancer for TestNextAttributes {
+    /// Advances the origin to the given block.
+    async fn advance_origin(&mut self) -> PipelineResult<()> {
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl NextAttributes for TestNextAttributes {
+    /// Returns the next valid attributes.
+    async fn next_attributes(
+        &mut self,
+        _: L2BlockInfo,
+    ) -> PipelineResult<OptimismAttributesWithParent> {
+        self.next_attributes.take().ok_or(PipelineError::Eof.temp())
+    }
+}
+
+/// An [L1Traversal] using test providers and sources.
+pub type TestL1Traversal = L1Traversal<TestChainProvider>;
+
+/// An [L1Retrieval] stage using test providers and sources.
+pub type TestL1Retrieval = L1Retrieval<TestDAP, TestL1Traversal>;
+
+/// A [FrameQueue] using test providers and sources.
+pub type TestFrameQueue = FrameQueue<TestL1Retrieval>;
+
+/// A [ChannelBank] using test providers and sources.
+pub type TestChannelBank = ChannelBank<TestFrameQueue>;
+
+/// A [ChannelReader] using test providers and sources.
+pub type TestChannelReader = ChannelReader<TestChannelBank>;
+
+/// A [BatchStream] using test providers and sources.
+pub type TestBatchStream = BatchStream<TestChannelReader, TestL2ChainProvider>;
+
+/// A [BatchQueue] using test providers and sources.
+pub type TestBatchQueue = BatchQueue<TestBatchStream, TestL2ChainProvider>;
+
+/// An [AttributesQueue] using test providers and sources.
+pub type TestAttributesQueue = AttributesQueue<TestBatchQueue, MockAttributesBuilder>;
+
+/// A [DerivationPipeline] using test providers and sources.
+pub type TestPipeline = DerivationPipeline<TestAttributesQueue, TestL2ChainProvider>;
+
+/// Constructs a [DerivationPipeline] using test providers and sources.
+pub fn new_test_pipeline() -> TestPipeline {
+    PipelineBuilder::new()
+        .rollup_config(Arc::new(RollupConfig::default()))
+        .origin(BlockInfo::default())
+        .dap_source(TestDAP::default())
+        .builder(MockAttributesBuilder::default())
+        .chain_provider(TestChainProvider::default())
+        .l2_chain_provider(TestL2ChainProvider::default())
+        .build()
+}

crates/derive/src/traits/pipeline.rs

@@ -9,7 +9,7 @@ use op_alloy_protocol::{BlockInfo, L2BlockInfo};
 use op_alloy_rpc_types_engine::OptimismAttributesWithParent;
 
 /// A pipeline error.
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Eq)]
 pub enum StepResult {
     /// Attributes were successfully prepared.
     PreparedAttributes,