Add unit tests for the equivocation detection loop (#2571)

* Add unit tests for the equivocation detection loop

* clippy

* use std::future::pending()

bridges/relays/equivocation/src/block_checker.rs

@@ -28,6 +28,7 @@ use num_traits::Saturating;
 ///
 /// Getting the finality info associated to the source headers synced with the target chain
 /// at the specified block.
+#[cfg_attr(test, derive(Debug, PartialEq))]
 pub struct ReadSyncedHeaders<P: EquivocationDetectionPipeline> {
 	pub target_block_num: P::TargetNumber,
 }
@@ -61,6 +62,7 @@ impl<P: EquivocationDetectionPipeline> ReadSyncedHeaders<P> {
 /// Second step in the block checking state machine.
 ///
 /// Reading the equivocation reporting context from the target chain.
+#[cfg_attr(test, derive(Debug))]
 pub struct ReadContext<P: EquivocationDetectionPipeline> {
 	target_block_num: P::TargetNumber,
 	synced_headers: Vec<HeaderFinalityInfo<P>>,
@@ -104,6 +106,7 @@ impl<P: EquivocationDetectionPipeline> ReadContext<P> {
 /// Third step in the block checking state machine.
 ///
 /// Searching for equivocations in the source headers synced with the target chain.
+#[cfg_attr(test, derive(Debug))]
 pub struct FindEquivocations<P: EquivocationDetectionPipeline> {
 	target_block_num: P::TargetNumber,
 	synced_headers: Vec<HeaderFinalityInfo<P>>,
@@ -122,10 +125,13 @@ impl<P: EquivocationDetectionPipeline> FindEquivocations<P> {
 				&synced_header.finality_proof,
 				finality_proofs_buf.buf().as_slice(),
 			) {
-				Ok(equivocations) => result.push(ReportEquivocations {
-					source_block_hash: self.context.synced_header_hash,
-					equivocations,
-				}),
+				Ok(equivocations) =>
+					if !equivocations.is_empty() {
+						result.push(ReportEquivocations {
+							source_block_hash: self.context.synced_header_hash,
+							equivocations,
+						})
+					},
 				Err(e) => {
 					log::error!(
 						target: "bridge",
@@ -148,6 +154,7 @@ impl<P: EquivocationDetectionPipeline> FindEquivocations<P> {
 /// Fourth step in the block checking state machine.
 ///
 /// Reporting the detected equivocations (if any).
+#[cfg_attr(test, derive(Debug))]
 pub struct ReportEquivocations<P: EquivocationDetectionPipeline> {
 	source_block_hash: P::Hash,
 	equivocations: Vec<P::EquivocationProof>,
@@ -157,7 +164,7 @@ impl<P: EquivocationDetectionPipeline> ReportEquivocations<P> {
 	pub async fn next<SC: SourceClient<P>>(
 		mut self,
 		source_client: &mut SC,
-		reporter: &mut EquivocationsReporter<P, SC>,
+		reporter: &mut EquivocationsReporter<'_, P, SC>,
 	) -> Result<(), Self> {
 		let mut unprocessed_equivocations = vec![];
 		for equivocation in self.equivocations {
@@ -191,6 +198,7 @@ impl<P: EquivocationDetectionPipeline> ReportEquivocations<P> {
 }
 
 /// Block checking state machine.
+#[cfg_attr(test, derive(Debug))]
 pub enum BlockChecker<P: EquivocationDetectionPipeline> {
 	ReadSyncedHeaders(ReadSyncedHeaders<P>),
 	ReadContext(ReadContext<P>),
@@ -250,3 +258,214 @@ impl<P: EquivocationDetectionPipeline> BlockChecker<P> {
 		.boxed()
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::mock::*;
+	use std::collections::HashMap;
+
+	impl PartialEq for ReadContext<TestEquivocationDetectionPipeline> {
+		fn eq(&self, other: &Self) -> bool {
+			self.target_block_num == other.target_block_num &&
+				self.synced_headers == other.synced_headers
+		}
+	}
+
+	impl PartialEq for FindEquivocations<TestEquivocationDetectionPipeline> {
+		fn eq(&self, other: &Self) -> bool {
+			self.target_block_num == other.target_block_num &&
+				self.synced_headers == other.synced_headers &&
+				self.context == other.context
+		}
+	}
+
+	impl PartialEq for ReportEquivocations<TestEquivocationDetectionPipeline> {
+		fn eq(&self, other: &Self) -> bool {
+			self.source_block_hash == other.source_block_hash &&
+				self.equivocations == other.equivocations
+		}
+	}
+
+	impl PartialEq for BlockChecker<TestEquivocationDetectionPipeline> {
+		fn eq(&self, _other: &Self) -> bool {
+			matches!(self, _other)
+		}
+	}
+
+	#[async_std::test]
+	async fn block_checker_works() {
+		let mut source_client = TestSourceClient { ..Default::default() };
+		let mut target_client = TestTargetClient {
+			best_synced_header_hash: HashMap::from([(9, Ok(Some(5)))]),
+			finality_verification_context: HashMap::from([(
+				9,
+				Ok(TestFinalityVerificationContext { check_equivocations: true }),
+			)]),
+			synced_headers_finality_info: HashMap::from([(
+				10,
+				Ok(vec![
+					new_header_finality_info(6, None),
+					new_header_finality_info(7, Some(false)),
+					new_header_finality_info(8, None),
+					new_header_finality_info(9, Some(true)),
+					new_header_finality_info(10, None),
+					new_header_finality_info(11, None),
+					new_header_finality_info(12, None),
+				]),
+			)]),
+			..Default::default()
+		};
+		let mut reporter =
+			EquivocationsReporter::<TestEquivocationDetectionPipeline, TestSourceClient>::new();
+
+		let block_checker = BlockChecker::new(10);
+		assert!(block_checker
+			.run(
+				&mut source_client,
+				&mut target_client,
+				&mut FinalityProofsBuf::new(vec![
+					TestFinalityProof(6, vec!["6-1"]),
+					TestFinalityProof(7, vec![]),
+					TestFinalityProof(8, vec!["8-1"]),
+					TestFinalityProof(9, vec!["9-1"]),
+					TestFinalityProof(10, vec![]),
+					TestFinalityProof(11, vec!["11-1", "11-2"]),
+					TestFinalityProof(12, vec!["12-1"])
+				]),
+				&mut reporter
+			)
+			.await
+			.is_ok());
+		assert_eq!(
+			*source_client.reported_equivocations.lock().unwrap(),
+			HashMap::from([(5, vec!["6-1"]), (9, vec!["11-1", "11-2", "12-1"])])
+		);
+	}
+
+	#[async_std::test]
+	async fn block_checker_works_with_empty_context() {
+		let mut target_client = TestTargetClient {
+			best_synced_header_hash: HashMap::from([(9, Ok(None))]),
+			finality_verification_context: HashMap::from([(
+				9,
+				Ok(TestFinalityVerificationContext { check_equivocations: true }),
+			)]),
+			synced_headers_finality_info: HashMap::from([(
+				10,
+				Ok(vec![new_header_finality_info(6, None)]),
+			)]),
+			..Default::default()
+		};
+		let mut source_client = TestSourceClient { ..Default::default() };
+		let mut reporter =
+			EquivocationsReporter::<TestEquivocationDetectionPipeline, TestSourceClient>::new();
+
+		let block_checker = BlockChecker::new(10);
+		assert!(block_checker
+			.run(
+				&mut source_client,
+				&mut target_client,
+				&mut FinalityProofsBuf::new(vec![TestFinalityProof(6, vec!["6-1"])]),
+				&mut reporter
+			)
+			.await
+			.is_ok());
+		assert_eq!(*source_client.reported_equivocations.lock().unwrap(), HashMap::default());
+	}
+
+	#[async_std::test]
+	async fn read_synced_headers_handles_errors() {
+		let mut target_client = TestTargetClient {
+			synced_headers_finality_info: HashMap::from([
+				(10, Err(TestClientError::NonConnection)),
+				(11, Err(TestClientError::Connection)),
+			]),
+			..Default::default()
+		};
+		let mut source_client = TestSourceClient { ..Default::default() };
+		let mut reporter =
+			EquivocationsReporter::<TestEquivocationDetectionPipeline, TestSourceClient>::new();
+
+		// NonConnection error
+		let block_checker = BlockChecker::new(10);
+		assert_eq!(
+			block_checker
+				.run(
+					&mut source_client,
+					&mut target_client,
+					&mut FinalityProofsBuf::new(vec![]),
+					&mut reporter
+				)
+				.await,
+			Err(BlockChecker::ReadSyncedHeaders(ReadSyncedHeaders { target_block_num: 10 }))
+		);
+		assert_eq!(target_client.num_reconnects, 0);
+
+		// Connection error
+		let block_checker = BlockChecker::new(11);
+		assert_eq!(
+			block_checker
+				.run(
+					&mut source_client,
+					&mut target_client,
+					&mut FinalityProofsBuf::new(vec![]),
+					&mut reporter
+				)
+				.await,
+			Err(BlockChecker::ReadSyncedHeaders(ReadSyncedHeaders { target_block_num: 11 }))
+		);
+		assert_eq!(target_client.num_reconnects, 1);
+	}
+
+	#[async_std::test]
+	async fn read_context_handles_errors() {
+		let mut target_client = TestTargetClient {
+			synced_headers_finality_info: HashMap::from([(10, Ok(vec![])), (11, Ok(vec![]))]),
+			best_synced_header_hash: HashMap::from([
+				(9, Err(TestClientError::NonConnection)),
+				(10, Err(TestClientError::Connection)),
+			]),
+			..Default::default()
+		};
+		let mut source_client = TestSourceClient { ..Default::default() };
+		let mut reporter =
+			EquivocationsReporter::<TestEquivocationDetectionPipeline, TestSourceClient>::new();
+
+		// NonConnection error
+		let block_checker = BlockChecker::new(10);
+		assert_eq!(
+			block_checker
+				.run(
+					&mut source_client,
+					&mut target_client,
+					&mut FinalityProofsBuf::new(vec![]),
+					&mut reporter
+				)
+				.await,
+			Err(BlockChecker::ReadContext(ReadContext {
+				target_block_num: 10,
+				synced_headers: vec![]
+			}))
+		);
+		assert_eq!(target_client.num_reconnects, 0);
+
+		// Connection error
+		let block_checker = BlockChecker::new(11);
+		assert_eq!(
+			block_checker
+				.run(
+					&mut source_client,
+					&mut target_client,
+					&mut FinalityProofsBuf::new(vec![]),
+					&mut reporter
+				)
+				.await,
+			Err(BlockChecker::ReadContext(ReadContext {
+				target_block_num: 11,
+				synced_headers: vec![]
+			}))
+		);
+		assert_eq!(target_client.num_reconnects, 1);
+	}
+}

bridges/relays/equivocation/src/equivocation_loop.rs

@@ -21,7 +21,7 @@ use crate::{
 
 use crate::block_checker::BlockChecker;
 use finality_relay::{FinalityProofsBuf, FinalityProofsStream};
-use futures::{select, FutureExt};
+use futures::{select_biased, FutureExt};
 use num_traits::Saturating;
 use relay_utils::{metrics::MetricsParams, FailedClient};
 use std::{future::Future, time::Duration};
@@ -38,7 +38,7 @@ struct EquivocationDetectionLoop<
 	from_block_num: Option<P::TargetNumber>,
 	until_block_num: Option<P::TargetNumber>,
 
-	reporter: EquivocationsReporter<P, SC>,
+	reporter: EquivocationsReporter<'static, P, SC>,
 
 	finality_proofs_stream: FinalityProofsStream<P, SC>,
 	finality_proofs_buf: FinalityProofsBuf<P>,
@@ -116,11 +116,11 @@ impl<P: EquivocationDetectionPipeline, SC: SourceClient<P>, TC: TargetClient<P>>
 					.await;
 				current_block_number = current_block_number.saturating_add(1.into());
 			}
-			self.until_block_num = Some(current_block_number);
+			self.from_block_num = Some(current_block_number);
 
-			select! {
-				_ = async_std::task::sleep(tick).fuse() => {},
+			select_biased! {
 				_ = exit_signal => return,
+				_ = async_std::task::sleep(tick).fuse() => {},
 			}
 		}
 	}
@@ -172,3 +172,137 @@ pub async fn run<P: EquivocationDetectionPipeline>(
 		)
 		.await
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::mock::*;
+	use futures::{channel::mpsc::UnboundedSender, StreamExt};
+	use std::{
+		collections::{HashMap, VecDeque},
+		sync::{Arc, Mutex},
+	};
+
+	fn best_finalized_header_number(
+		best_finalized_headers: &Mutex<VecDeque<Result<TestTargetNumber, TestClientError>>>,
+		exit_sender: &UnboundedSender<()>,
+	) -> Result<TestTargetNumber, TestClientError> {
+		let mut best_finalized_headers = best_finalized_headers.lock().unwrap();
+		let result = best_finalized_headers.pop_front().unwrap();
+		if best_finalized_headers.is_empty() {
+			exit_sender.unbounded_send(()).unwrap();
+		}
+		result
+	}
+
+	#[async_std::test]
+	async fn multiple_blocks_are_checked_correctly() {
+		let best_finalized_headers = Arc::new(Mutex::new(VecDeque::from([Ok(10), Ok(12), Ok(13)])));
+		let (exit_sender, exit_receiver) = futures::channel::mpsc::unbounded();
+
+		let source_client = TestSourceClient {
+			finality_proofs: Arc::new(Mutex::new(vec![
+				TestFinalityProof(2, vec!["2-1"]),
+				TestFinalityProof(3, vec!["3-1", "3-2"]),
+				TestFinalityProof(4, vec!["4-1"]),
+				TestFinalityProof(5, vec!["5-1"]),
+				TestFinalityProof(6, vec!["6-1", "6-2"]),
+				TestFinalityProof(7, vec!["7-1", "7-2"]),
+			])),
+			..Default::default()
+		};
+		let reported_equivocations = source_client.reported_equivocations.clone();
+		let target_client = TestTargetClient {
+			best_finalized_header_number: Arc::new(move || {
+				best_finalized_header_number(&best_finalized_headers, &exit_sender)
+			}),
+			best_synced_header_hash: HashMap::from([
+				(9, Ok(Some(1))),
+				(10, Ok(Some(3))),
+				(11, Ok(Some(5))),
+				(12, Ok(Some(6))),
+			]),
+			finality_verification_context: HashMap::from([
+				(9, Ok(TestFinalityVerificationContext { check_equivocations: true })),
+				(10, Ok(TestFinalityVerificationContext { check_equivocations: true })),
+				(11, Ok(TestFinalityVerificationContext { check_equivocations: false })),
+				(12, Ok(TestFinalityVerificationContext { check_equivocations: true })),
+			]),
+			synced_headers_finality_info: HashMap::from([
+				(
+					10,
+					Ok(vec![new_header_finality_info(2, None), new_header_finality_info(3, None)]),
+				),
+				(
+					11,
+					Ok(vec![
+						new_header_finality_info(4, None),
+						new_header_finality_info(5, Some(false)),
+					]),
+				),
+				(12, Ok(vec![new_header_finality_info(6, None)])),
+				(13, Ok(vec![new_header_finality_info(7, None)])),
+			]),
+			..Default::default()
+		};
+
+		assert!(run::<TestEquivocationDetectionPipeline>(
+			source_client,
+			target_client,
+			Duration::from_secs(0),
+			MetricsParams { address: None, registry: Default::default() },
+			exit_receiver.into_future().map(|(_, _)| ()),
+		)
+		.await
+		.is_ok());
+		assert_eq!(
+			*reported_equivocations.lock().unwrap(),
+			HashMap::from([
+				(1, vec!["2-1", "3-1", "3-2"]),
+				(3, vec!["4-1", "5-1"]),
+				(6, vec!["7-1", "7-2"])
+			])
+		);
+	}
+
+	#[async_std::test]
+	async fn blocks_following_error_are_checked_correctly() {
+		let best_finalized_headers = Mutex::new(VecDeque::from([Ok(10), Ok(11)]));
+		let (exit_sender, exit_receiver) = futures::channel::mpsc::unbounded();
+
+		let source_client = TestSourceClient {
+			finality_proofs: Arc::new(Mutex::new(vec![
+				TestFinalityProof(2, vec!["2-1"]),
+				TestFinalityProof(3, vec!["3-1"]),
+			])),
+			..Default::default()
+		};
+		let reported_equivocations = source_client.reported_equivocations.clone();
+		let target_client = TestTargetClient {
+			best_finalized_header_number: Arc::new(move || {
+				best_finalized_header_number(&best_finalized_headers, &exit_sender)
+			}),
+			best_synced_header_hash: HashMap::from([(9, Ok(Some(1))), (10, Ok(Some(2)))]),
+			finality_verification_context: HashMap::from([
+				(9, Ok(TestFinalityVerificationContext { check_equivocations: true })),
+				(10, Ok(TestFinalityVerificationContext { check_equivocations: true })),
+			]),
+			synced_headers_finality_info: HashMap::from([
+				(10, Err(TestClientError::NonConnection)),
+				(11, Ok(vec![new_header_finality_info(3, None)])),
+			]),
+			..Default::default()
+		};
+
+		assert!(run::<TestEquivocationDetectionPipeline>(
+			source_client,
+			target_client,
+			Duration::from_secs(0),
+			MetricsParams { address: None, registry: Default::default() },
+			exit_receiver.into_future().map(|(_, _)| ()),
+		)
+		.await
+		.is_ok());
+		assert_eq!(*reported_equivocations.lock().unwrap(), HashMap::from([(2, vec!["3-1"]),]));
+	}
+}

bridges/relays/equivocation/src/lib.rs

@@ -16,24 +16,27 @@
 
 mod block_checker;
 mod equivocation_loop;
+mod mock;
 mod reporter;
 
 use async_trait::async_trait;
 use bp_header_chain::{FinalityProof, FindEquivocations};
 use finality_relay::{FinalityPipeline, SourceClientBase};
-use relay_utils::{
-	relay_loop::{Client as RelayClient, RECONNECT_DELAY},
-	MaybeConnectionError, TransactionTracker,
-};
-use std::fmt::Debug;
+use relay_utils::{relay_loop::Client as RelayClient, MaybeConnectionError, TransactionTracker};
+use std::{fmt::Debug, time::Duration};
 
 pub use equivocation_loop::run;
 
+#[cfg(not(test))]
+const RECONNECT_DELAY: Duration = relay_utils::relay_loop::RECONNECT_DELAY;
+#[cfg(test)]
+const RECONNECT_DELAY: Duration = mock::TEST_RECONNECT_DELAY;
+
 pub trait EquivocationDetectionPipeline: FinalityPipeline {
 	/// Block number of the target chain.
 	type TargetNumber: relay_utils::BlockNumberBase;
 	/// The context needed for validating finality proofs.
-	type FinalityVerificationContext: Send;
+	type FinalityVerificationContext: Debug + Send;
 	/// The type of the equivocation proof.
 	type EquivocationProof: Clone + Debug + Send + Sync;
 	/// The equivocations finder.
@@ -91,6 +94,7 @@ pub trait TargetClient<P: EquivocationDetectionPipeline>: RelayClient {
 }
 
 /// The context needed for finding equivocations inside finality proofs and reporting them.
+#[derive(Debug, PartialEq)]
 struct EquivocationReportingContext<P: EquivocationDetectionPipeline> {
 	pub synced_header_hash: P::Hash,
 	pub synced_verification_context: P::FinalityVerificationContext,

bridges/relays/equivocation/src/mock.rs

@@ -0,0 +1,285 @@
+// Copyright (C) Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+#![cfg(test)]
+
+use crate::{EquivocationDetectionPipeline, HeaderFinalityInfo, SourceClient, TargetClient};
+use async_trait::async_trait;
+use bp_header_chain::{FinalityProof, FindEquivocations};
+use finality_relay::{FinalityPipeline, SourceClientBase};
+use futures::{Stream, StreamExt};
+use relay_utils::{
+	relay_loop::Client as RelayClient, HeaderId, MaybeConnectionError, TrackedTransactionStatus,
+	TransactionTracker,
+};
+use std::{
+	collections::HashMap,
+	pin::Pin,
+	sync::{Arc, Mutex},
+	time::Duration,
+};
+
+pub type TestSourceHashAndNumber = u64;
+pub type TestTargetNumber = u64;
+pub type TestEquivocationProof = &'static str;
+
+pub const TEST_RECONNECT_DELAY: Duration = Duration::from_secs(0);
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TestFinalityProof(pub TestSourceHashAndNumber, pub Vec<TestEquivocationProof>);
+
+impl FinalityProof<TestSourceHashAndNumber, TestSourceHashAndNumber> for TestFinalityProof {
+	fn target_header_hash(&self) -> TestSourceHashAndNumber {
+		self.0
+	}
+
+	fn target_header_number(&self) -> TestSourceHashAndNumber {
+		self.0
+	}
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct TestEquivocationDetectionPipeline;
+
+impl FinalityPipeline for TestEquivocationDetectionPipeline {
+	const SOURCE_NAME: &'static str = "TestSource";
+	const TARGET_NAME: &'static str = "TestTarget";
+
+	type Hash = TestSourceHashAndNumber;
+	type Number = TestSourceHashAndNumber;
+	type FinalityProof = TestFinalityProof;
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct TestFinalityVerificationContext {
+	pub check_equivocations: bool,
+}
+
+pub struct TestEquivocationsFinder;
+
+impl FindEquivocations<TestFinalityProof, TestFinalityVerificationContext, TestEquivocationProof>
+	for TestEquivocationsFinder
+{
+	type Error = ();
+
+	fn find_equivocations(
+		verification_context: &TestFinalityVerificationContext,
+		synced_proof: &TestFinalityProof,
+		source_proofs: &[TestFinalityProof],
+	) -> Result<Vec<TestEquivocationProof>, Self::Error> {
+		if verification_context.check_equivocations {
+			// Get the equivocations from the source proofs, in order to make sure
+			// that they are correctly provided.
+			if let Some(proof) = source_proofs.iter().find(|proof| proof.0 == synced_proof.0) {
+				return Ok(proof.1.clone())
+			}
+		}
+
+		Ok(vec![])
+	}
+}
+
+impl EquivocationDetectionPipeline for TestEquivocationDetectionPipeline {
+	type TargetNumber = TestTargetNumber;
+	type FinalityVerificationContext = TestFinalityVerificationContext;
+	type EquivocationProof = TestEquivocationProof;
+	type EquivocationsFinder = TestEquivocationsFinder;
+}
+
+#[derive(Debug, Clone)]
+pub enum TestClientError {
+	Connection,
+	NonConnection,
+}
+
+impl MaybeConnectionError for TestClientError {
+	fn is_connection_error(&self) -> bool {
+		match self {
+			TestClientError::Connection => true,
+			TestClientError::NonConnection => false,
+		}
+	}
+}
+
+#[derive(Clone)]
+pub struct TestSourceClient {
+	pub num_reconnects: u32,
+	pub finality_proofs: Arc<Mutex<Vec<TestFinalityProof>>>,
+	pub reported_equivocations:
+		Arc<Mutex<HashMap<TestSourceHashAndNumber, Vec<TestEquivocationProof>>>>,
+}
+
+impl Default for TestSourceClient {
+	fn default() -> Self {
+		Self {
+			num_reconnects: 0,
+			finality_proofs: Arc::new(Mutex::new(vec![])),
+			reported_equivocations: Arc::new(Mutex::new(Default::default())),
+		}
+	}
+}
+
+#[async_trait]
+impl RelayClient for TestSourceClient {
+	type Error = TestClientError;
+
+	async fn reconnect(&mut self) -> Result<(), Self::Error> {
+		self.num_reconnects += 1;
+
+		Ok(())
+	}
+}
+
+#[async_trait]
+impl SourceClientBase<TestEquivocationDetectionPipeline> for TestSourceClient {
+	type FinalityProofsStream = Pin<Box<dyn Stream<Item = TestFinalityProof> + 'static + Send>>;
+
+	async fn finality_proofs(&self) -> Result<Self::FinalityProofsStream, Self::Error> {
+		let finality_proofs = std::mem::take(&mut *self.finality_proofs.lock().unwrap());
+		Ok(futures::stream::iter(finality_proofs).boxed())
+	}
+}
+
+#[derive(Clone, Debug)]
+pub struct TestTransactionTracker(
+	pub TrackedTransactionStatus<HeaderId<TestSourceHashAndNumber, TestSourceHashAndNumber>>,
+);
+
+impl Default for TestTransactionTracker {
+	fn default() -> TestTransactionTracker {
+		TestTransactionTracker(TrackedTransactionStatus::Finalized(Default::default()))
+	}
+}
+
+#[async_trait]
+impl TransactionTracker for TestTransactionTracker {
+	type HeaderId = HeaderId<TestSourceHashAndNumber, TestSourceHashAndNumber>;
+
+	async fn wait(
+		self,
+	) -> TrackedTransactionStatus<HeaderId<TestSourceHashAndNumber, TestSourceHashAndNumber>> {
+		self.0
+	}
+}
+
+#[async_trait]
+impl SourceClient<TestEquivocationDetectionPipeline> for TestSourceClient {
+	type TransactionTracker = TestTransactionTracker;
+
+	async fn report_equivocation(
+		&self,
+		at: TestSourceHashAndNumber,
+		equivocation: TestEquivocationProof,
+	) -> Result<Self::TransactionTracker, Self::Error> {
+		self.reported_equivocations
+			.lock()
+			.unwrap()
+			.entry(at)
+			.or_default()
+			.push(equivocation);
+
+		Ok(TestTransactionTracker::default())
+	}
+}
+
+#[derive(Clone)]
+pub struct TestTargetClient {
+	pub num_reconnects: u32,
+	pub best_finalized_header_number:
+		Arc<dyn Fn() -> Result<TestTargetNumber, TestClientError> + Send + Sync>,
+	pub best_synced_header_hash:
+		HashMap<TestTargetNumber, Result<Option<TestSourceHashAndNumber>, TestClientError>>,
+	pub finality_verification_context:
+		HashMap<TestTargetNumber, Result<TestFinalityVerificationContext, TestClientError>>,
+	pub synced_headers_finality_info: HashMap<
+		TestTargetNumber,
+		Result<Vec<HeaderFinalityInfo<TestEquivocationDetectionPipeline>>, TestClientError>,
+	>,
+}
+
+impl Default for TestTargetClient {
+	fn default() -> Self {
+		Self {
+			num_reconnects: 0,
+			best_finalized_header_number: Arc::new(|| Ok(0)),
+			best_synced_header_hash: Default::default(),
+			finality_verification_context: Default::default(),
+			synced_headers_finality_info: Default::default(),
+		}
+	}
+}
+
+#[async_trait]
+impl RelayClient for TestTargetClient {
+	type Error = TestClientError;
+
+	async fn reconnect(&mut self) -> Result<(), Self::Error> {
+		self.num_reconnects += 1;
+
+		Ok(())
+	}
+}
+
+#[async_trait]
+impl TargetClient<TestEquivocationDetectionPipeline> for TestTargetClient {
+	async fn best_finalized_header_number(&self) -> Result<TestTargetNumber, Self::Error> {
+		(self.best_finalized_header_number)()
+	}
+
+	async fn best_synced_header_hash(
+		&self,
+		at: TestTargetNumber,
+	) -> Result<Option<TestSourceHashAndNumber>, Self::Error> {
+		self.best_synced_header_hash
+			.get(&at)
+			.unwrap_or(&Err(TestClientError::NonConnection))
+			.clone()
+	}
+
+	async fn finality_verification_context(
+		&self,
+		at: TestTargetNumber,
+	) -> Result<TestFinalityVerificationContext, Self::Error> {
+		self.finality_verification_context
+			.get(&at)
+			.unwrap_or(&Err(TestClientError::NonConnection))
+			.clone()
+	}
+
+	async fn synced_headers_finality_info(
+		&self,
+		at: TestTargetNumber,
+	) -> Result<Vec<HeaderFinalityInfo<TestEquivocationDetectionPipeline>>, Self::Error> {
+		self.synced_headers_finality_info
+			.get(&at)
+			.unwrap_or(&Err(TestClientError::NonConnection))
+			.clone()
+	}
+}
+
+pub fn new_header_finality_info(
+	source_hdr: TestSourceHashAndNumber,
+	check_following_equivocations: Option<bool>,
+) -> HeaderFinalityInfo<TestEquivocationDetectionPipeline> {
+	HeaderFinalityInfo::<TestEquivocationDetectionPipeline> {
+		finality_proof: TestFinalityProof(source_hdr, vec![]),
+		new_verification_context: check_following_equivocations.map(
+			|check_following_equivocations| TestFinalityVerificationContext {
+				check_equivocations: check_following_equivocations,
+			},
+		),
+	}
+}

bridges/relays/equivocation/src/reporter.rs

@@ -25,11 +25,11 @@ use std::{
 	task::{Context, Poll},
 };
 
-pub struct EquivocationsReporter<P: EquivocationDetectionPipeline, SC: SourceClient<P>> {
-	pending_reports: Vec<TrackedTransactionFuture<SC::TransactionTracker>>,
+pub struct EquivocationsReporter<'a, P: EquivocationDetectionPipeline, SC: SourceClient<P>> {
+	pending_reports: Vec<TrackedTransactionFuture<'a, SC::TransactionTracker>>,
 }
 
-impl<P: EquivocationDetectionPipeline, SC: SourceClient<P>> EquivocationsReporter<P, SC> {
+impl<'a, P: EquivocationDetectionPipeline, SC: SourceClient<P>> EquivocationsReporter<'a, P, SC> {
 	pub fn new() -> Self {
 		Self { pending_reports: vec![] }
 	}
@@ -81,3 +81,49 @@ impl<P: EquivocationDetectionPipeline, SC: SourceClient<P>> EquivocationsReporte
 		poll_fn(|cx| self.do_process_pending_reports(cx)).await
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::mock::*;
+	use relay_utils::HeaderId;
+	use std::sync::Mutex;
+
+	#[async_std::test]
+	async fn process_pending_reports_works() {
+		let polled_reports = Mutex::new(vec![]);
+		let finished_reports = Mutex::new(vec![]);
+
+		let mut reporter =
+			EquivocationsReporter::<TestEquivocationDetectionPipeline, TestSourceClient> {
+				pending_reports: vec![
+					Box::pin(async {
+						polled_reports.lock().unwrap().push(1);
+						finished_reports.lock().unwrap().push(1);
+						TrackedTransactionStatus::Finalized(HeaderId(1, 1))
+					}),
+					Box::pin(async {
+						polled_reports.lock().unwrap().push(2);
+						finished_reports.lock().unwrap().push(2);
+						TrackedTransactionStatus::Finalized(HeaderId(2, 2))
+					}),
+					Box::pin(async {
+						polled_reports.lock().unwrap().push(3);
+						std::future::pending::<()>().await;
+						finished_reports.lock().unwrap().push(3);
+						TrackedTransactionStatus::Finalized(HeaderId(3, 3))
+					}),
+					Box::pin(async {
+						polled_reports.lock().unwrap().push(4);
+						finished_reports.lock().unwrap().push(4);
+						TrackedTransactionStatus::Finalized(HeaderId(4, 4))
+					}),
+				],
+			};
+
+		reporter.process_pending_reports().await;
+		assert_eq!(*polled_reports.lock().unwrap(), vec![1, 2, 3, 4]);
+		assert_eq!(*finished_reports.lock().unwrap(), vec![1, 2, 4]);
+		assert_eq!(reporter.pending_reports.len(), 1);
+	}
+}