Restore integration tests (#2620)

* restored test_consensus

* restored test_sync

* run integration tests only on CI

* use single test-thread for integration tests

* post-merge fix

* panic when integration test runs for too long

* add some traces to integration logs

* manual wait -> timeout

* post-merge fix

* post-merge fix

substrate/.gitlab-ci.yml

@@ -96,6 +96,34 @@ test-linux-stable:                 &test
     - time cargo test --all --release --verbose --locked
     - sccache -s
 
+test-linux-stable-int:             &test
+  stage:                           test
+  variables:
+    RUST_TOOLCHAIN: stable
+    # Enable debug assertions since we are running optimized builds for testing
+    # but still want to have debug assertions.
+    RUSTFLAGS: -Cdebug-assertions=y
+    TARGET: native
+  tags:
+    - linux-docker
+  only:
+    - tags
+    - master
+    - schedules
+    - web
+    - /^[0-9]+$/
+  except:
+    variables:
+      - $DEPLOY_TAG
+  before_script:
+    - sccache -s
+    - ./scripts/build.sh
+  script:
+    - time RUST_LOG=sync=trace,consensus=trace,client=trace,state-db=trace,db=trace,forks=trace,state_db=trace,storage_cache=trace cargo test -p node-cli --release --verbose --locked -- --ignored --test-threads=1
+    - sccache -s
+  allow_failure:                   true
+
+
 check-web-wasm:
   stage:                           test
   <<:                              *compiler_info

substrate/Cargo.lock

@@ -2152,6 +2152,7 @@ dependencies = [
  "substrate-cli 2.0.0",
  "substrate-client 2.0.0",
  "substrate-consensus-aura 2.0.0",
+ "substrate-consensus-common 2.0.0",
  "substrate-finality-grandpa 2.0.0",
  "substrate-inherents 2.0.0",
  "substrate-keyring 2.0.0",

substrate/core/consensus/aura/src/lib.rs

@@ -69,9 +69,9 @@ use slots::{CheckedHeader, SlotWorker, SlotInfo, SlotCompatible, slot_now, check
 
 pub use aura_primitives::*;
 pub use consensus_common::{SyncOracle, ExtraVerification};
+pub use digest::CompatibleDigestItem;
 
 mod digest;
-use digest::CompatibleDigestItem;
 
 type AuthorityId<P> = <P as Pair>::Public;
 

substrate/core/service/test/src/lib.rs

@@ -24,7 +24,7 @@ use std::collections::HashMap;
 use log::info;
 use futures::{Future, Stream};
 use tempdir::TempDir;
-use tokio::runtime::Runtime;
+use tokio::{runtime::Runtime, prelude::FutureExt};
 use tokio::timer::Interval;
 use service::{
 	ServiceFactory,
@@ -39,6 +39,9 @@ use network::config::{NetworkConfiguration, NodeKeyConfig, Secret, NonReservedPe
 use sr_primitives::generic::BlockId;
 use consensus::{ImportBlock, BlockImport};
 
+/// Maximum duration of single wait call.
+const MAX_WAIT_TIME: Duration = Duration::from_secs(60 * 3);
+
 struct TestNet<F: ServiceFactory> {
 	runtime: Runtime,
 	authority_nodes: Vec<(u32, Arc<F::FullService>, Multiaddr)>,
@@ -52,14 +55,22 @@ struct TestNet<F: ServiceFactory> {
 impl<F: ServiceFactory> TestNet<F> {
 	pub fn run_until_all_full<P: Send + Sync + Fn(u32, &F::FullService) -> bool + 'static>(&mut self, predicate: P) {
 		let full_nodes = self.full_nodes.clone();
-		let interval = Interval::new_interval(Duration::from_millis(100)).map_err(|_| ()).for_each(move |_| {
-			if full_nodes.iter().all(|&(ref id, ref service, _)| predicate(*id, service)) {
-				Err(())
-			} else {
-				Ok(())
-			}
-		});
-		self.runtime.block_on(interval).ok();
+		let interval = Interval::new_interval(Duration::from_millis(100))
+			.map_err(|_| ())
+			.for_each(move |_| {
+				if full_nodes.iter().all(|&(ref id, ref service, _)| predicate(*id, service)) {
+					Err(())
+				} else {
+					Ok(())
+				}
+			})
+			.timeout(MAX_WAIT_TIME);
+
+		match self.runtime.block_on(interval) {
+			Ok(()) => unreachable!("interval always fails; qed"),
+			Err(ref err) if err.is_inner() => (),
+			Err(_) => panic!("Waited for too long"),
+		}
 	}
 }
 
@@ -223,11 +234,15 @@ pub fn connectivity<F: ServiceFactory>(spec: FactoryChainSpec<F>) {
 	}
 }
 
-pub fn sync<F, B, E>(spec: FactoryChainSpec<F>, block_factory: B, extrinsic_factory: E)
+pub fn sync<F, B, E>(
+	spec: FactoryChainSpec<F>,
+	mut block_factory: B,
+	mut extrinsic_factory: E,
+)
 where
 	F: ServiceFactory,
-	B: Fn(&F::FullService) -> ImportBlock<F::Block>,
-	E: Fn(&F::FullService) -> FactoryExtrinsic<F>,
+	B: FnMut(&F::FullService) -> ImportBlock<F::Block>,
+	E: FnMut(&F::FullService) -> FactoryExtrinsic<F>,
 {
 	const NUM_NODES: u32 = 10;
 	const NUM_BLOCKS: u32 = 512;
@@ -265,8 +280,8 @@ pub fn consensus<F>(spec: FactoryChainSpec<F>, authorities: Vec<String>)
 	where
 		F: ServiceFactory,
 {
-	const NUM_NODES: u32 = 20;
-	const NUM_BLOCKS: u32 = 200;
+	const NUM_NODES: u32 = 10;
+	const NUM_BLOCKS: u32 = 10; // 10 * 2 sec block production time = ~20 seconds
 	let temp = TempDir::new("substrate-conensus-test").expect("Error creating test dir");
 	let mut network = TestNet::<F>::new(&temp, spec.clone(), NUM_NODES / 2, 0, authorities, 30600);
 	info!("Checking consensus");

substrate/node/cli/Cargo.toml

@@ -39,6 +39,7 @@ rand = "0.6"
 finality_tracker = { package = "srml-finality-tracker", path = "../../srml/finality-tracker", default-features = false }
 
 [dev-dependencies]
+consensus-common = { package = "substrate-consensus-common", path = "../../core/consensus/common" }
 service-test = { package = "substrate-service-test", path = "../../core/service/test" }
 
 [build-dependencies]

substrate/node/cli/src/chain_spec.rs

@@ -370,7 +370,7 @@ pub fn local_testnet_config() -> ChainSpec {
 }
 
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
 	use super::*;
 	use service_test;
 	use crate::service::Factory;
@@ -381,13 +381,41 @@ mod tests {
 		genesis
 	}
 
+	fn local_testnet_genesis_instant_single() -> GenesisConfig {
+		let mut genesis = testnet_genesis(
+			vec![
+				get_authority_keys_from_seed("Alice"),
+			],
+			get_account_id_from_seed("Alice"),
+			None,
+			false,
+		);
+		genesis.timestamp = Some(TimestampConfig { minimum_period: 1 });
+		genesis
+	}
+
+	/// Local testnet config (single validator - Alice)
+	pub fn integration_test_config_with_single_authority() -> ChainSpec {
+		ChainSpec::from_genesis(
+			"Integration Test",
+			"test",
+			local_testnet_genesis_instant_single,
+			vec![],
+			None,
+			None,
+			None,
+			None,
+		)
+	}
+
 	/// Local testnet config (multivalidator Alice + Bob)
-	pub fn integration_test_config() -> ChainSpec {
+	pub fn integration_test_config_with_two_authorities() -> ChainSpec {
 		ChainSpec::from_genesis("Integration Test", "test", local_testnet_genesis_instant, vec![], None, None, None, None)
 	}
 
 	#[test]
+	#[ignore]
 	fn test_connectivity() {
-		service_test::connectivity::<Factory>(integration_test_config());
+		service_test::connectivity::<Factory>(integration_test_config_with_two_authorities());
 	}
 }

substrate/node/cli/src/service.rs

@@ -220,6 +220,23 @@ construct_service_factory! {
 
 #[cfg(test)]
 mod tests {
+	use std::sync::Arc;
+	use consensus::CompatibleDigestItem;
+	use consensus_common::{Environment, Proposer, ImportBlock, BlockOrigin, ForkChoiceStrategy};
+	use node_primitives::DigestItem;
+	use node_runtime::{Call, BalancesCall, UncheckedExtrinsic};
+	use parity_codec::{Compact, Encode, Decode};
+	use primitives::{
+		crypto::Pair as CryptoPair, ed25519::Pair, blake2_256,
+		sr25519::Public as AddressPublic,
+	};
+	use sr_primitives::{generic::{BlockId, Era, Digest}, traits::{Block, Digest as DigestT}, OpaqueExtrinsic};
+	use timestamp;
+	use finality_tracker;
+	use keyring::{ed25519::Keyring as AuthorityKeyring, sr25519::Keyring as AccountKeyring};
+	use substrate_service::ServiceFactory;
+	use crate::service::Factory;
+
 	#[cfg(feature = "rhd")]
 	fn test_sync() {
 		use {service_test, Factory};
@@ -267,4 +284,107 @@ mod tests {
 		service_test::sync::<Factory, _, _>(chain_spec::integration_test_config(), block_factory, extrinsic_factory);
 	}
 
+	#[test]
+	#[ignore]
+	fn test_sync() {
+		let chain_spec = crate::chain_spec::tests::integration_test_config_with_single_authority();
+
+		let alice = Arc::new(AuthorityKeyring::Alice.pair());
+		let mut slot_num = 1u64;
+		let block_factory = |service: &<Factory as ServiceFactory>::FullService| {
+			let mut inherent_data = service.config.custom.inherent_data_providers
+				.create_inherent_data().unwrap();
+			inherent_data.replace_data(finality_tracker::INHERENT_IDENTIFIER, &1u64);
+			inherent_data.replace_data(timestamp::INHERENT_IDENTIFIER, &(slot_num * 10));
+
+			let parent_id = BlockId::number(service.client().info().unwrap().chain.best_number);
+			let parent_header = service.client().header(&parent_id).unwrap().unwrap();
+			let proposer_factory = Arc::new(substrate_basic_authorship::ProposerFactory {
+				client: service.client(),
+				transaction_pool: service.transaction_pool(),
+			});
+			let mut digest = Digest::<DigestItem>::default();
+			digest.push(<DigestItem as CompatibleDigestItem<Pair>>::aura_pre_digest(slot_num * 10 / 2));
+			let proposer = proposer_factory.init(&parent_header, &[]).unwrap();
+			let new_block = proposer.propose(
+				inherent_data,
+				digest,
+				::std::time::Duration::from_secs(1),
+			).expect("Error making test block");
+
+			let (new_header, new_body) = new_block.deconstruct();
+			let pre_hash = new_header.hash();
+			// sign the pre-sealed hash of the block and then
+			// add it to a digest item.
+			let to_sign = pre_hash.encode();
+			let signature = alice.sign(&to_sign[..]);
+			let item = <DigestItem as CompatibleDigestItem<Pair>>::aura_seal(
+				signature,
+			);
+			slot_num += 1;
+
+			ImportBlock {
+				origin: BlockOrigin::File,
+				header: new_header,
+				justification: None,
+				post_digests: vec![item],
+				body: Some(new_body),
+				finalized: true,
+				auxiliary: Vec::new(),
+				fork_choice: ForkChoiceStrategy::LongestChain,
+			}
+		};
+
+		let bob = Arc::new(AccountKeyring::Bob.pair());
+		let charlie = Arc::new(AccountKeyring::Charlie.pair());
+
+		let mut index = 0;
+		let extrinsic_factory = |service: &<Factory as ServiceFactory>::FullService| {
+			let amount = 1000;
+			let to = AddressPublic::from_raw(bob.public().0);
+			let from = AddressPublic::from_raw(charlie.public().0);
+			let genesis_hash = service.client().block_hash(0).unwrap().unwrap();
+			let signer = charlie.clone();
+
+			let function = Call::Balances(BalancesCall::transfer(to.into(), amount));
+			let era = Era::immortal();
+			let raw_payload = (Compact(index), function, era, genesis_hash);
+			let signature = raw_payload.using_encoded(|payload| if payload.len() > 256 {
+				signer.sign(&blake2_256(payload)[..])
+			} else {
+				signer.sign(payload)
+			});
+			let xt = UncheckedExtrinsic::new_signed(
+				index,
+				raw_payload.1,
+				from.into(),
+				signature.into(),
+				era,
+			).encode();
+			let v: Vec<u8> = Decode::decode(&mut xt.as_slice()).unwrap();
+
+			index += 1;
+			OpaqueExtrinsic(v)
+		};
+
+		service_test::sync::<Factory, _, _>(
+			chain_spec,
+			block_factory,
+			extrinsic_factory,
+		);
+	}
+
+	#[test]
+	#[ignore]
+	fn test_consensus() {
+		use super::Factory;
+
+		service_test::consensus::<Factory>(
+			crate::chain_spec::tests::integration_test_config_with_two_authorities(),
+			vec![
+				"//Alice".into(),
+				"//Bob".into(),
+			],
+		)
+	}
 }

substrate/node/primitives/src/lib.rs

@@ -57,9 +57,10 @@ pub type Hash = primitives::H256;
 /// A timestamp: seconds since the unix epoch.
 pub type Timestamp = u64;
 
+/// Digest item type.
+pub type DigestItem = generic::DigestItem<Hash, AuthorityId, AuthoritySignature>;
 /// Header type.
-/// 
-pub type Header = generic::Header<BlockNumber, BlakeTwo256, generic::DigestItem<Hash, AuthorityId, AuthoritySignature>>;
+pub type Header = generic::Header<BlockNumber, BlakeTwo256, DigestItem>;
 /// Block type.
 pub type Block = generic::Block<Header, UncheckedExtrinsic>;
 /// Block ID.