diff --git a/Cargo.lock b/Cargo.lock
index a00f0da12..56ee426f4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -328,6 +328,7 @@ dependencies = [
  "async-trait",
  "hashbrown",
  "serde",
+ "tokio",
  "unsigned-varint",
 ]
 
diff --git a/crates/derive/Cargo.toml b/crates/derive/Cargo.toml
index 9f26eaf77..38e640b31 100644
--- a/crates/derive/Cargo.toml
+++ b/crates/derive/Cargo.toml
@@ -23,5 +23,8 @@ unsigned-varint = "0.8.0"
 # Optional
 serde = { version = "1.0.197", default-features = false, features = ["derive"], optional = true }
 
+[dev-dependencies]
+tokio = { version = "1.36", features = ["full"] }
+
 [features]
 serde = ["dep:serde", "alloy-primitives/serde"]
diff --git a/crates/derive/src/stages/channel_bank.rs b/crates/derive/src/stages/channel_bank.rs
index 6293a6d14..c872c9ce1 100644
--- a/crates/derive/src/stages/channel_bank.rs
+++ b/crates/derive/src/stages/channel_bank.rs
@@ -60,12 +60,15 @@ where
         self.prev.origin()
     }
 
+    /// Returns the size of the channel bank by accumulating over all channels.
+    pub fn size(&self) -> usize {
+        self.channels.iter().fold(0, |acc, (_, c)| acc + c.size())
+    }
+
     /// Prunes the Channel bank, until it is below [MAX_CHANNEL_BANK_SIZE].
+    /// Prunes from the high-priority channel since it failed to be read.
     pub fn prune(&mut self) -> StageResult<()> {
-        // Check total size
-        let mut total_size = self.channels.iter().fold(0, |acc, (_, c)| acc + c.size());
-        // Prune until it is reasonable again. The high-priority channel failed to be read,
-        // so we prune from there.
+        let mut total_size = self.size();
         while total_size > MAX_CHANNEL_BANK_SIZE {
             let id = self
                 .channel_queue
@@ -122,16 +125,17 @@ where
             .ok_or(anyhow!("Channel not found"))?;
         let origin = self.origin().ok_or(anyhow!("No origin present"))?;
 
+        // Remove all timed out channels from the front of the `channel_queue`.
         if channel.open_block_number() + self.cfg.channel_timeout < origin.number {
             self.channels.remove(&first);
             self.channel_queue.pop_front();
             return Ok(None);
         }
 
-        // At the point we have removed all timed out channels from the front of the `channel_queue`.
+        // At this point we have removed all timed out channels from the front of the `channel_queue`.
         // Pre-Canyon we simply check the first index.
-        // Post-Canyon we read the entire channelQueue for the first ready channel. If no channel is
-        // available, we return `nil, io.EOF`.
+        // Post-Canyon we read the entire channelQueue for the first ready channel.
+        // If no channel is available, we return StageError::Eof.
         // Canyon is activated when the first L1 block whose time >= CanyonTime, not on the L2 timestamp.
         if !self.cfg.is_canyon_active(origin.timestamp) {
             return self.try_read_channel_at_index(0).map(Some);
@@ -201,3 +205,68 @@ where
         Err(StageError::Eof)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::stages::frame_queue::tests::new_test_frames;
+    use crate::stages::l1_retrieval::L1Retrieval;
+    use crate::stages::l1_traversal::tests::new_test_traversal;
+    use crate::traits::test_utils::TestDAP;
+    use alloc::vec;
+
+    #[test]
+    fn test_ingest_empty_origin() {
+        let mut traversal = new_test_traversal(false, false);
+        traversal.block = None;
+        let dap = TestDAP::default();
+        let retrieval = L1Retrieval::new(traversal, dap);
+        let frame_queue = FrameQueue::new(retrieval);
+        let mut channel_bank = ChannelBank::new(RollupConfig::default(), frame_queue);
+        let frame = Frame::default();
+        let err = channel_bank.ingest_frame(frame).unwrap_err();
+        assert_eq!(err, StageError::Custom(anyhow!("No origin")));
+    }
+
+    #[test]
+    fn test_ingest_and_prune_channel_bank() {
+        let traversal = new_test_traversal(true, true);
+        let results = vec![Ok(Bytes::from(vec![0x00]))];
+        let dap = TestDAP { results };
+        let retrieval = L1Retrieval::new(traversal, dap);
+        let frame_queue = FrameQueue::new(retrieval);
+        let mut channel_bank = ChannelBank::new(RollupConfig::default(), frame_queue);
+        let mut frames = new_test_frames(100000);
+        // Ingest frames until the channel bank is full and it stops increasing in size
+        let mut current_size = 0;
+        let next_frame = frames.pop().unwrap();
+        channel_bank.ingest_frame(next_frame).unwrap();
+        while channel_bank.size() > current_size {
+            current_size = channel_bank.size();
+            let next_frame = frames.pop().unwrap();
+            channel_bank.ingest_frame(next_frame).unwrap();
+            assert!(channel_bank.size() <= MAX_CHANNEL_BANK_SIZE);
+        }
+        // There should be a bunch of frames leftover
+        assert!(!frames.is_empty());
+        // If we ingest one more frame, the channel bank should prune
+        // and the size should be the same
+        let next_frame = frames.pop().unwrap();
+        channel_bank.ingest_frame(next_frame).unwrap();
+        assert_eq!(channel_bank.size(), current_size);
+    }
+
+    #[tokio::test]
+    async fn test_read_empty_channel_bank() {
+        let traversal = new_test_traversal(true, true);
+        let results = vec![Ok(Bytes::from(vec![0x00]))];
+        let dap = TestDAP { results };
+        let retrieval = L1Retrieval::new(traversal, dap);
+        let frame_queue = FrameQueue::new(retrieval);
+        let mut channel_bank = ChannelBank::new(RollupConfig::default(), frame_queue);
+        let err = channel_bank.read().unwrap_err();
+        assert_eq!(err, StageError::Eof);
+        let err = channel_bank.next_data().await.unwrap_err();
+        assert_eq!(err, StageError::Custom(anyhow!("Not Enough Data")));
+    }
+}
diff --git a/crates/derive/src/stages/frame_queue.rs b/crates/derive/src/stages/frame_queue.rs
index f0900a91b..51221eb63 100644
--- a/crates/derive/src/stages/frame_queue.rs
+++ b/crates/derive/src/stages/frame_queue.rs
@@ -47,6 +47,7 @@ where
         if self.queue.is_empty() {
             match self.prev.next_data().await {
                 Ok(data) => {
+                    // TODO: what do we do with frame parsing errors?
                     if let Ok(frames) = Frame::parse_frames(data.as_ref()) {
                         self.queue.extend(frames);
                     }
@@ -78,3 +79,112 @@ where
         Err(StageError::Eof)
     }
 }
+
+#[cfg(test)]
+pub(crate) mod tests {
+    use super::*;
+    use crate::stages::l1_traversal::tests::new_test_traversal;
+    use crate::traits::test_utils::TestDAP;
+    use crate::DERIVATION_VERSION_0;
+    use alloc::vec;
+    use alloc::vec::Vec;
+    use alloy_primitives::Bytes;
+
+    pub(crate) fn new_test_frames(count: usize) -> Vec<Frame> {
+        (0..count)
+            .map(|i| Frame {
+                id: [0xFF; 16],
+                number: i as u16,
+                data: vec![0xDD; 50],
+                is_last: i == count - 1,
+            })
+            .collect()
+    }
+
+    pub(crate) fn new_encoded_test_frames(count: usize) -> Bytes {
+        let frames = new_test_frames(count);
+        let mut bytes = Vec::new();
+        bytes.extend_from_slice(&[DERIVATION_VERSION_0]);
+        for frame in frames.iter() {
+            bytes.extend_from_slice(&frame.encode());
+        }
+        Bytes::from(bytes)
+    }
+
+    #[tokio::test]
+    async fn test_frame_queue_empty_bytes() {
+        let traversal = new_test_traversal(true, true);
+        let results = vec![Ok(Bytes::from(vec![0x00]))];
+        let dap = TestDAP { results };
+        let retrieval = L1Retrieval::new(traversal, dap);
+        let mut frame_queue = FrameQueue::new(retrieval);
+        let err = frame_queue.next_frame().await.unwrap_err();
+        assert_eq!(err, anyhow!("Not enough data").into());
+    }
+
+    #[tokio::test]
+    async fn test_frame_queue_no_frames_decoded() {
+        let traversal = new_test_traversal(true, true);
+        let results = vec![Err(StageError::Eof), Ok(Bytes::default())];
+        let dap = TestDAP { results };
+        let retrieval = L1Retrieval::new(traversal, dap);
+        let mut frame_queue = FrameQueue::new(retrieval);
+        let err = frame_queue.next_frame().await.unwrap_err();
+        assert_eq!(err, anyhow!("Not enough data").into());
+    }
+
+    #[tokio::test]
+    async fn test_frame_queue_wrong_derivation_version() {
+        let traversal = new_test_traversal(true, true);
+        let results = vec![Ok(Bytes::from(vec![0x01]))];
+        let dap = TestDAP { results };
+        let retrieval = L1Retrieval::new(traversal, dap);
+        let mut frame_queue = FrameQueue::new(retrieval);
+        let err = frame_queue.next_frame().await.unwrap_err();
+        assert_eq!(err, anyhow!("Unsupported derivation version").into());
+    }
+
+    #[tokio::test]
+    async fn test_frame_queue_frame_too_short() {
+        let traversal = new_test_traversal(true, true);
+        let results = vec![Ok(Bytes::from(vec![0x00, 0x01]))];
+        let dap = TestDAP { results };
+        let retrieval = L1Retrieval::new(traversal, dap);
+        let mut frame_queue = FrameQueue::new(retrieval);
+        let err = frame_queue.next_frame().await.unwrap_err();
+        assert_eq!(err, anyhow!("Frame too short to decode").into());
+    }
+
+    #[tokio::test]
+    async fn test_frame_queue_single_frame() {
+        let data = new_encoded_test_frames(1);
+        let traversal = new_test_traversal(true, true);
+        let dap = TestDAP {
+            results: vec![Ok(data)],
+        };
+        let retrieval = L1Retrieval::new(traversal, dap);
+        let mut frame_queue = FrameQueue::new(retrieval);
+        let frame_decoded = frame_queue.next_frame().await.unwrap();
+        let frame = new_test_frames(1);
+        assert_eq!(frame[0], frame_decoded);
+        let err = frame_queue.next_frame().await.unwrap_err();
+        assert_eq!(err, anyhow!("Not enough data").into());
+    }
+
+    #[tokio::test]
+    async fn test_frame_queue_multiple_frames() {
+        let data = new_encoded_test_frames(3);
+        let traversal = new_test_traversal(true, true);
+        let dap = TestDAP {
+            results: vec![Ok(data)],
+        };
+        let retrieval = L1Retrieval::new(traversal, dap);
+        let mut frame_queue = FrameQueue::new(retrieval);
+        for i in 0..3 {
+            let frame_decoded = frame_queue.next_frame().await.unwrap();
+            assert_eq!(frame_decoded.number, i);
+        }
+        let err = frame_queue.next_frame().await.unwrap_err();
+        assert_eq!(err, anyhow!("Not enough data").into());
+    }
+}
diff --git a/crates/derive/src/stages/l1_retrieval.rs b/crates/derive/src/stages/l1_retrieval.rs
index ca4ecc5b9..c69dea9c3 100644
--- a/crates/derive/src/stages/l1_retrieval.rs
+++ b/crates/derive/src/stages/l1_retrieval.rs
@@ -22,7 +22,7 @@ where
     /// The data availability provider to use for the L1 retrieval stage.
     pub provider: DAP,
     /// The current data iterator.
-    data: Option<DAP::DataIter<Bytes>>,
+    pub(crate) data: Option<DAP::DataIter>,
 }
 
 impl<DAP, CP> L1Retrieval<DAP, CP>
@@ -83,3 +83,82 @@ where
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::stages::l1_traversal::tests::new_test_traversal;
+    use crate::traits::test_utils::{TestDAP, TestIter};
+    use alloc::vec;
+    use alloy_primitives::Address;
+
+    #[tokio::test]
+    async fn test_l1_retrieval_origin() {
+        let traversal = new_test_traversal(true, true);
+        let dap = TestDAP { results: vec![] };
+        let retrieval = L1Retrieval::new(traversal, dap);
+        let expected = BlockInfo::default();
+        assert_eq!(retrieval.origin(), Some(&expected));
+    }
+
+    #[tokio::test]
+    async fn test_l1_retrieval_next_data() {
+        let traversal = new_test_traversal(true, true);
+        let results = vec![Err(StageError::Eof), Ok(Bytes::default())];
+        let dap = TestDAP { results };
+        let mut retrieval = L1Retrieval::new(traversal, dap);
+        assert_eq!(retrieval.data, None);
+        let data = retrieval.next_data().await.unwrap();
+        assert_eq!(data, Bytes::default());
+        assert!(retrieval.data.is_some());
+        let retrieval_data = retrieval.data.as_ref().unwrap();
+        assert_eq!(retrieval_data.open_data_calls.len(), 1);
+        assert_eq!(retrieval_data.open_data_calls[0].0, BlockInfo::default());
+        assert_eq!(retrieval_data.open_data_calls[0].1, Address::default());
+        // Data should be reset to none and the error should be bubbled up.
+        let data = retrieval.next_data().await.unwrap_err();
+        assert_eq!(data, StageError::Eof);
+        assert!(retrieval.data.is_none());
+    }
+
+    #[tokio::test]
+    async fn test_l1_retrieval_existing_data_is_respected() {
+        let data = TestIter {
+            open_data_calls: vec![(BlockInfo::default(), Address::default())],
+            results: vec![Ok(Bytes::default())],
+        };
+        // Create a new traversal with no blocks or receipts.
+        // This would bubble up an error if the prev stage
+        // (traversal) is called in the retrieval stage.
+        let traversal = new_test_traversal(false, false);
+        let dap = TestDAP { results: vec![] };
+        let mut retrieval = L1Retrieval {
+            prev: traversal,
+            provider: dap,
+            data: Some(data),
+        };
+        let data = retrieval.next_data().await.unwrap();
+        assert_eq!(data, Bytes::default());
+        assert!(retrieval.data.is_some());
+        let retrieval_data = retrieval.data.as_ref().unwrap();
+        assert_eq!(retrieval_data.open_data_calls.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn test_l1_retrieval_existing_data_errors() {
+        let data = TestIter {
+            open_data_calls: vec![(BlockInfo::default(), Address::default())],
+            results: vec![Err(StageError::Eof)],
+        };
+        let traversal = new_test_traversal(true, true);
+        let dap = TestDAP { results: vec![] };
+        let mut retrieval = L1Retrieval {
+            prev: traversal,
+            provider: dap,
+            data: Some(data),
+        };
+        let data = retrieval.next_data().await.unwrap_err();
+        assert_eq!(data, StageError::Eof);
+        assert!(retrieval.data.is_none());
+    }
+}
diff --git a/crates/derive/src/stages/l1_traversal.rs b/crates/derive/src/stages/l1_traversal.rs
index 253611050..37656ca49 100644
--- a/crates/derive/src/stages/l1_traversal.rs
+++ b/crates/derive/src/stages/l1_traversal.rs
@@ -12,7 +12,7 @@ use async_trait::async_trait;
 #[derive(Debug, Clone, Copy)]
 pub struct L1Traversal<Provider: ChainProvider> {
     /// The current block in the traversal stage.
-    block: Option<BlockInfo>,
+    pub(crate) block: Option<BlockInfo>,
     /// The data source for the traversal stage.
     data_source: Provider,
     /// Signals whether or not the traversal stage has been completed.
@@ -27,7 +27,7 @@ impl<F: ChainProvider> L1Traversal<F> {
     /// Creates a new [L1Traversal] instance.
     pub fn new(data_source: F, cfg: RollupConfig) -> Self {
         Self {
-            block: None,
+            block: Some(BlockInfo::default()),
             data_source,
             done: false,
             system_config: SystemConfig::default(),
@@ -35,8 +35,14 @@ impl<F: ChainProvider> L1Traversal<F> {
         }
     }
 
-    /// Returns the next L1 block in the traversal stage, if the stage has not been completed. This function can only
-    /// be called once, and will return `None` on subsequent calls unless the stage is reset.
+    /// Retrieves a reference to the inner data source of the [L1Traversal] stage.
+    pub fn data_source(&self) -> &F {
+        &self.data_source
+    }
+
+    /// Returns the next L1 block in the traversal stage, if the stage has not been completed.
+    /// This function can only be called once, and will return `None` on subsequent calls
+    /// unless the stage is reset.
     pub fn next_l1_block(&mut self) -> StageResult<Option<BlockInfo>> {
         if !self.done {
             self.done = true;
@@ -53,8 +59,9 @@ impl<F: ChainProvider> L1Traversal<F> {
 
     /// Advances the internal state of the [L1Traversal] stage to the next L1 block.
     pub async fn advance_l1_block(&mut self) -> StageResult<()> {
-        // TODO: Return EOF if the block wasn't found.
-        let block = self.block.ok_or(anyhow!("No block to advance from"))?;
+        // Pull the next block or return EOF which has special
+        // handling further up the pipeline.
+        let block = self.block.ok_or(StageError::Eof)?;
         let next_l1_origin = self
             .data_source
             .block_info_by_number(block.number + 1)
@@ -96,3 +103,114 @@ impl<F: ChainProvider + Send> ResettableStage for L1Traversal<F> {
         Err(StageError::Eof)
     }
 }
+
+#[cfg(test)]
+pub(crate) mod tests {
+    use super::*;
+    use crate::traits::test_utils::TestChainProvider;
+    use crate::types::{Receipt, CONFIG_UPDATE_EVENT_VERSION_0, CONFIG_UPDATE_TOPIC};
+    use alloc::vec;
+    use alloy_primitives::{address, b256, hex, Address, Bytes, Log, LogData, B256};
+
+    const L1_SYS_CONFIG_ADDR: Address = address!("1337000000000000000000000000000000000000");
+
+    fn new_update_batcher_log() -> Log {
+        const UPDATE_TYPE: B256 =
+            b256!("0000000000000000000000000000000000000000000000000000000000000000");
+        Log {
+            address: L1_SYS_CONFIG_ADDR,
+            data: LogData::new_unchecked(
+                vec![
+                    CONFIG_UPDATE_TOPIC,
+                    CONFIG_UPDATE_EVENT_VERSION_0,
+                    UPDATE_TYPE,
+                ],
+                hex!("00000000000000000000000000000000000000000000000000000000000000200000000000000000000000000000000000000000000000000000000000000020000000000000000000000000000000000000000000000000000000000000beef").into()
+            )
+        }
+    }
+
+    pub(crate) fn new_test_traversal(
+        blocks: bool,
+        receipts: bool,
+    ) -> L1Traversal<TestChainProvider> {
+        let mut provider = TestChainProvider::default();
+        let rollup_config = RollupConfig {
+            l1_system_config_address: L1_SYS_CONFIG_ADDR,
+            ..RollupConfig::default()
+        };
+        let block = BlockInfo::default();
+        if blocks {
+            provider.insert_block(0, block);
+            provider.insert_block(1, block);
+        }
+        if receipts {
+            let mut receipt = Receipt {
+                success: true,
+                ..Receipt::default()
+            };
+            let bad = Log::new(
+                Address::from([2; 20]),
+                vec![CONFIG_UPDATE_TOPIC, B256::default()],
+                Bytes::default(),
+            )
+            .unwrap();
+            receipt.logs = vec![new_update_batcher_log(), bad, new_update_batcher_log()];
+            let receipts = vec![receipt.clone(), Receipt::default(), receipt];
+            provider.insert_receipts(block.hash, receipts);
+        }
+        L1Traversal::new(provider, rollup_config)
+    }
+
+    #[tokio::test]
+    async fn test_l1_traversal() {
+        let mut traversal = new_test_traversal(true, true);
+        assert_eq!(
+            traversal.next_l1_block().unwrap(),
+            Some(BlockInfo::default())
+        );
+        assert_eq!(traversal.next_l1_block().unwrap_err(), StageError::Eof);
+        assert!(traversal.advance_l1_block().await.is_ok());
+    }
+
+    #[tokio::test]
+    async fn test_l1_traversal_missing_receipts() {
+        let mut traversal = new_test_traversal(true, false);
+        assert_eq!(
+            traversal.next_l1_block().unwrap(),
+            Some(BlockInfo::default())
+        );
+        assert_eq!(traversal.next_l1_block().unwrap_err(), StageError::Eof);
+        matches!(
+            traversal.advance_l1_block().await.unwrap_err(),
+            StageError::Custom(_)
+        );
+    }
+
+    #[tokio::test]
+    async fn test_l1_traversal_missing_blocks() {
+        let mut traversal = new_test_traversal(false, false);
+        assert_eq!(
+            traversal.next_l1_block().unwrap(),
+            Some(BlockInfo::default())
+        );
+        assert_eq!(traversal.next_l1_block().unwrap_err(), StageError::Eof);
+        matches!(
+            traversal.advance_l1_block().await.unwrap_err(),
+            StageError::Custom(_)
+        );
+    }
+
+    #[tokio::test]
+    async fn test_system_config_updated() {
+        let mut traversal = new_test_traversal(true, true);
+        assert_eq!(
+            traversal.next_l1_block().unwrap(),
+            Some(BlockInfo::default())
+        );
+        assert_eq!(traversal.next_l1_block().unwrap_err(), StageError::Eof);
+        assert!(traversal.advance_l1_block().await.is_ok());
+        let expected = address!("000000000000000000000000000000000000bEEF");
+        assert_eq!(traversal.system_config.batcher_addr, expected);
+    }
+}
diff --git a/crates/derive/src/traits/data_sources.rs b/crates/derive/src/traits/data_sources.rs
index 69ccee18e..a1521ce21 100644
--- a/crates/derive/src/traits/data_sources.rs
+++ b/crates/derive/src/traits/data_sources.rs
@@ -21,16 +21,16 @@ pub trait ChainProvider {
 /// Describes the functionality of a data source that can provide data availability information.
 #[async_trait]
 pub trait DataAvailabilityProvider {
-    /// A data iterator for the data source to return.
-    type DataIter<T: Into<Bytes>>: DataIter<T> + Send + Debug;
+    /// An iterator over returned bytes data.
+    type DataIter: DataIter<Bytes> + Send + Debug;
 
     /// Returns the data availability for the block with the given hash, or an error if the block does not exist in the
     /// data source.
-    async fn open_data<T: Into<Bytes>>(
+    async fn open_data(
         &self,
         block_ref: &BlockInfo,
         batcher_address: Address,
-    ) -> Result<Self::DataIter<T>>;
+    ) -> Result<Self::DataIter>;
 }
 
 /// Describes the behavior of a data iterator.
diff --git a/crates/derive/src/traits/mod.rs b/crates/derive/src/traits/mod.rs
index a1c2b46fe..cd1d1ab9f 100644
--- a/crates/derive/src/traits/mod.rs
+++ b/crates/derive/src/traits/mod.rs
@@ -5,3 +5,6 @@ pub use data_sources::{ChainProvider, DataAvailabilityProvider, DataIter};
 
 mod stages;
 pub use stages::ResettableStage;
+
+#[cfg(test)]
+pub mod test_utils;
diff --git a/crates/derive/src/traits/test_utils.rs b/crates/derive/src/traits/test_utils.rs
new file mode 100644
index 000000000..94eaa6cbc
--- /dev/null
+++ b/crates/derive/src/traits/test_utils.rs
@@ -0,0 +1,7 @@
+//! Test Utilities for derive traits
+
+pub mod data_sources;
+pub use data_sources::TestChainProvider;
+
+pub mod data_availability;
+pub use data_availability::{TestDAP, TestIter};
diff --git a/crates/derive/src/traits/test_utils/data_availability.rs b/crates/derive/src/traits/test_utils/data_availability.rs
new file mode 100644
index 000000000..9f21378ea
--- /dev/null
+++ b/crates/derive/src/traits/test_utils/data_availability.rs
@@ -0,0 +1,58 @@
+//! Test utilities for data availability.
+
+use crate::{
+    traits::{DataAvailabilityProvider, DataIter},
+    types::{BlockInfo, StageError, StageResult},
+};
+use alloc::{boxed::Box, vec, vec::Vec};
+use alloy_primitives::{Address, Bytes};
+use anyhow::Result;
+use async_trait::async_trait;
+use core::fmt::Debug;
+
+/// Mock data iterator
+#[derive(Debug, Default, PartialEq)]
+pub struct TestIter {
+    /// Holds open data calls with args for assertions.
+    pub(crate) open_data_calls: Vec<(BlockInfo, Address)>,
+    /// A queue of results to return as the next iterated data.
+    pub(crate) results: Vec<StageResult<Bytes>>,
+}
+
+impl DataIter<Bytes> for TestIter {
+    fn next(&mut self) -> StageResult<Bytes> {
+        self.results.pop().unwrap_or_else(|| Err(StageError::Eof))
+    }
+}
+
+/// Mock data availability provider
+#[derive(Debug, Default)]
+pub struct TestDAP {
+    /// Specifies the stage results the test iter returns as data.
+    pub(crate) results: Vec<StageResult<Bytes>>,
+}
+
+#[async_trait]
+impl DataAvailabilityProvider for TestDAP {
+    type DataIter = TestIter;
+
+    async fn open_data(
+        &self,
+        block_ref: &BlockInfo,
+        batcher_address: Address,
+    ) -> Result<Self::DataIter> {
+        // Construct a new vec of results to return.
+        let results = self
+            .results
+            .iter()
+            .map(|i| match i {
+                Ok(r) => Ok(r.clone()),
+                Err(_) => Err(StageError::Eof),
+            })
+            .collect::<Vec<StageResult<Bytes>>>();
+        Ok(TestIter {
+            open_data_calls: vec![(*block_ref, batcher_address)],
+            results,
+        })
+    }
+}
diff --git a/crates/derive/src/traits/test_utils/data_sources.rs b/crates/derive/src/traits/test_utils/data_sources.rs
new file mode 100644
index 000000000..02518b44e
--- /dev/null
+++ b/crates/derive/src/traits/test_utils/data_sources.rs
@@ -0,0 +1,64 @@
+//! Data Sources Test Utilities
+
+use crate::traits::ChainProvider;
+use crate::types::{BlockInfo, Receipt};
+use alloc::{boxed::Box, vec::Vec};
+use alloy_primitives::B256;
+use anyhow::Result;
+use async_trait::async_trait;
+
+/// A mock chain provider for testing.
+#[derive(Debug, Clone, Default)]
+pub struct TestChainProvider {
+    /// Maps block numbers to block information using a tuple list.
+    pub blocks: Vec<(u64, BlockInfo)>,
+    /// Maps block hashes to receipts using a tuple list.
+    pub receipts: Vec<(B256, Vec<Receipt>)>,
+}
+
+impl TestChainProvider {
+    /// Insert a block into the mock chain provider.
+    pub fn insert_block(&mut self, number: u64, block: BlockInfo) {
+        self.blocks.push((number, block));
+    }
+
+    /// Insert receipts into the mock chain provider.
+    pub fn insert_receipts(&mut self, hash: B256, receipts: Vec<Receipt>) {
+        self.receipts.push((hash, receipts));
+    }
+
+    /// Clears blocks from the mock chain provider.
+    pub fn clear_blocks(&mut self) {
+        self.blocks.clear();
+    }
+
+    /// Clears receipts from the mock chain provider.
+    pub fn clear_receipts(&mut self) {
+        self.receipts.clear();
+    }
+
+    /// Clears all blocks and receipts from the mock chain provider.
+    pub fn clear(&mut self) {
+        self.clear_blocks();
+        self.clear_receipts();
+    }
+}
+
+#[async_trait]
+impl ChainProvider for TestChainProvider {
+    async fn block_info_by_number(&self, _number: u64) -> Result<BlockInfo> {
+        if let Some((_, block)) = self.blocks.iter().find(|(n, _)| *n == _number) {
+            Ok(*block)
+        } else {
+            Err(anyhow::anyhow!("Block not found"))
+        }
+    }
+
+    async fn receipts_by_hash(&self, _hash: B256) -> Result<Vec<Receipt>> {
+        if let Some((_, receipts)) = self.receipts.iter().find(|(h, _)| *h == _hash) {
+            Ok(receipts.clone())
+        } else {
+            Err(anyhow::anyhow!("Receipts not found"))
+        }
+    }
+}
diff --git a/crates/derive/src/types/block.rs b/crates/derive/src/types/block.rs
index 67543c7ab..36012fc2c 100644
--- a/crates/derive/src/types/block.rs
+++ b/crates/derive/src/types/block.rs
@@ -56,6 +56,12 @@ pub enum BlockId {
     Kind(BlockKind),
 }
 
+impl Default for BlockId {
+    fn default() -> Self {
+        BlockId::Kind(BlockKind::Latest)
+    }
+}
+
 /// The Block Kind
 ///
 /// The block kinds are:
diff --git a/crates/derive/src/types/errors.rs b/crates/derive/src/types/errors.rs
index 7d34b8404..98d3220b8 100644
--- a/crates/derive/src/types/errors.rs
+++ b/crates/derive/src/types/errors.rs
@@ -14,6 +14,17 @@ pub enum StageError {
     Custom(anyhow::Error),
 }
 
+impl PartialEq<StageError> for StageError {
+    fn eq(&self, other: &StageError) -> bool {
+        matches!(
+            (self, other),
+            (StageError::Eof, StageError::Eof)
+                | (StageError::NotEnoughData, StageError::NotEnoughData)
+                | (StageError::Custom(_), StageError::Custom(_))
+        )
+    }
+}
+
 /// A result type for the derivation pipeline stages.
 pub type StageResult<T> = Result<T, StageError>;
 
diff --git a/crates/derive/src/types/genesis.rs b/crates/derive/src/types/genesis.rs
index 4acd3f562..671efc3c4 100644
--- a/crates/derive/src/types/genesis.rs
+++ b/crates/derive/src/types/genesis.rs
@@ -3,7 +3,7 @@
 use super::{BlockId, SystemConfig};
 
 /// Represents the genesis state of the rollup.
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, Default)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub struct Genesis {
     /// The L1 block that the rollup starts *after* (no derived transactions)
diff --git a/crates/derive/src/types/mod.rs b/crates/derive/src/types/mod.rs
index da8a53720..91b7725cb 100644
--- a/crates/derive/src/types/mod.rs
+++ b/crates/derive/src/types/mod.rs
@@ -4,7 +4,10 @@ use alloc::vec::Vec;
 use alloy_rlp::{Decodable, Encodable};
 
 mod system_config;
-pub use system_config::{SystemAccounts, SystemConfig, SystemConfigUpdateType};
+pub use system_config::{
+    SystemAccounts, SystemConfig, SystemConfigUpdateType, CONFIG_UPDATE_EVENT_VERSION_0,
+    CONFIG_UPDATE_TOPIC,
+};
 
 mod rollup_config;
 pub use rollup_config::RollupConfig;
diff --git a/crates/derive/src/types/rollup_config.rs b/crates/derive/src/types/rollup_config.rs
index c43e8bd1b..d1501b4ff 100644
--- a/crates/derive/src/types/rollup_config.rs
+++ b/crates/derive/src/types/rollup_config.rs
@@ -4,7 +4,7 @@ use super::Genesis;
 use alloy_primitives::Address;
 
 /// The Rollup configuration.
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, Default)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub struct RollupConfig {
     /// The genesis state of the rollup.
diff --git a/crates/derive/src/types/system_config.rs b/crates/derive/src/types/system_config.rs
index 0775cc8cb..0ddd85862 100644
--- a/crates/derive/src/types/system_config.rs
+++ b/crates/derive/src/types/system_config.rs
@@ -6,11 +6,11 @@ use alloy_sol_types::{sol, SolType};
 use anyhow::{anyhow, bail, Result};
 
 /// `keccak256("ConfigUpdate(uint256,uint8,bytes)")`
-const CONFIG_UPDATE_TOPIC: B256 =
+pub const CONFIG_UPDATE_TOPIC: B256 =
     b256!("1d2b0bda21d56b8bd12d4f94ebacffdfb35f5e226f84b461103bb8beab6353be");
 
 /// The initial version of the system config event log.
-const CONFIG_UPDATE_EVENT_VERSION_0: B256 = B256::ZERO;
+pub const CONFIG_UPDATE_EVENT_VERSION_0: B256 = B256::ZERO;
 
 /// Optimism system config contract values
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]