Additional fetch tests (#3983)

* First bunch of tests

* Dapps zip tests

dapps/src/tests/helpers/fetch.rs

@@ -14,21 +14,71 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
-use std::{io, thread};
-use std::sync::Arc;
-use std::sync::atomic::{self, AtomicUsize};
+use std::{io, thread, time};
+use std::sync::{atomic, mpsc, Arc};
 use util::Mutex;
 
 use futures::{self, Future};
 use fetch::{self, Fetch};
 
+pub struct FetchControl {
+	sender: mpsc::Sender<()>,
+	fetch: FakeFetch,
+}
+
+impl FetchControl {
+	pub fn respond(self) {
+		self.sender.send(())
+			.expect("Fetch cannot be finished without sending a response at least once.");
+	}
+
+	pub fn wait_for_requests(&self, len: usize) {
+		const MAX_TIMEOUT_MS: u64 = 5000;
+		const ATTEMPTS: u64 = 10;
+		let mut attempts_left = ATTEMPTS;
+		loop {
+			let current = self.fetch.requested.lock().len();
+
+			if current == len {
+				break;
+			} else if attempts_left == 0 {
+				panic!(
+					"Timeout reached when waiting for pending requests. Expected: {}, current: {}",
+					len, current
+				);
+			} else {
+				attempts_left -= 1;
+				// Should we handle spurious timeouts better?
+				thread::park_timeout(time::Duration::from_millis(MAX_TIMEOUT_MS / ATTEMPTS));
+			}
+		}
+	}
+}
+
 #[derive(Clone, Default)]
 pub struct FakeFetch {
-	asserted: Arc<AtomicUsize>,
+	manual: Arc<Mutex<Option<mpsc::Receiver<()>>>>,
+	response: Arc<Mutex<Option<&'static [u8]>>>,
+	asserted: Arc<atomic::AtomicUsize>,
 	requested: Arc<Mutex<Vec<String>>>,
 }
 
 impl FakeFetch {
+	pub fn set_response(&self, data: &'static [u8]) {
+		*self.response.lock() = Some(data);
+	}
+
+	pub fn manual(&self) -> FetchControl {
+		assert!(self.manual.lock().is_none(), "Only one manual control may be active.");
+		let (tx, rx) = mpsc::channel();
+		*self.manual.lock() = Some(rx);
+
+		FetchControl {
+			sender: tx,
+			fetch: self.clone(),
+		}
+	}
+
 	pub fn assert_requested(&self, url: &str) {
 		let requests = self.requested.lock();
 		let idx = self.asserted.fetch_add(1, atomic::Ordering::SeqCst);
@@ -52,10 +102,18 @@ impl Fetch for FakeFetch {
 
 	fn fetch_with_abort(&self, url: &str, _abort: fetch::Abort) -> Self::Result {
 		self.requested.lock().push(url.into());
+		let manual = self.manual.clone();
+		let response = self.response.clone();
 
 		let (tx, rx) = futures::oneshot();
 		thread::spawn(move || {
-			let cursor = io::Cursor::new(b"Some content");
+			if let Some(rx) = manual.lock().take() {
+				// wait for manual resume
+				let _ = rx.recv();
+			}
+
+			let data = response.lock().take().unwrap_or(b"Some content");
+			let cursor = io::Cursor::new(data);
 			tx.complete(fetch::Response::from_reader(cursor));
 		});
 

dapps/src/tests/helpers/mod.rs

@@ -42,7 +42,7 @@ fn init_logger() {
 	}
 }
 
-pub fn init_server<F, B>(hosts: Option<Vec<String>>, process: F) -> (Server, Arc<FakeRegistrar>) where
+pub fn init_server<F, B>(hosts: Option<Vec<String>>, process: F, remote: Remote) -> (Server, Arc<FakeRegistrar>) where
 	F: FnOnce(ServerBuilder) -> ServerBuilder<B>,
 	B: Fetch,
 {
@@ -51,7 +51,7 @@ pub fn init_server<F, B>(hosts: Option<Vec<String>>, process: F) -> (Server, Arc
 	let mut dapps_path = env::temp_dir();
 	dapps_path.push("non-existent-dir-to-prevent-fs-files-from-loading");
 	let server = process(ServerBuilder::new(
-		dapps_path.to_str().unwrap().into(), registrar.clone(), Remote::new_sync()
+		dapps_path.to_str().unwrap().into(), registrar.clone(), remote,
 	))
 		.signer_address(Some(("127.0.0.1".into(), SIGNER_PORT)))
 		.start_unsecured_http(&"127.0.0.1:0".parse().unwrap(), hosts).unwrap();
@@ -72,25 +72,29 @@ pub fn serve_with_auth(user: &str, pass: &str) -> Server {
 }
 
 pub fn serve_hosts(hosts: Option<Vec<String>>) -> Server {
-	init_server(hosts, |builder| builder).0
+	init_server(hosts, |builder| builder, Remote::new_sync()).0
 }
 
 pub fn serve_with_registrar() -> (Server, Arc<FakeRegistrar>) {
-	init_server(None, |builder| builder)
+	init_server(None, |builder| builder, Remote::new_sync())
 }
 
 pub fn serve_with_registrar_and_sync() -> (Server, Arc<FakeRegistrar>) {
 	init_server(None, |builder| {
 		builder.sync_status(Arc::new(|| true))
-	})
+	}, Remote::new_sync())
 }
 
 pub fn serve_with_registrar_and_fetch() -> (Server, FakeFetch, Arc<FakeRegistrar>) {
+	serve_with_registrar_and_fetch_and_threads(false)
+}
+
+pub fn serve_with_registrar_and_fetch_and_threads(multi_threaded: bool) -> (Server, FakeFetch, Arc<FakeRegistrar>) {
 	let fetch = FakeFetch::default();
 	let f = fetch.clone();
 	let (server, reg) = init_server(None, move |builder| {
 		builder.fetch(f.clone())
-	});
+	}, if multi_threaded { Remote::new_thread_per_future() } else { Remote::new_sync() });
 
 	(server, fetch, reg)
 }
@@ -102,13 +106,13 @@ pub fn serve_with_fetch(web_token: &'static str) -> (Server, FakeFetch) {
 		builder
 			.fetch(f.clone())
 			.web_proxy_tokens(Arc::new(move |token| &token == web_token))
-	});
+	}, Remote::new_sync());
 
 	(server, fetch)
 }
 
 pub fn serve() -> Server {
-	init_server(None, |builder| builder).0
+	init_server(None, |builder| builder, Remote::new_sync()).0
 }
 
 pub fn request(server: Server, request: &str) -> http_client::Response {