Refactor file notification monitoring in tests

Create a `PathWatcher` helper type that presents itself as an `Iterator`
over events on a specified path. It also allows setting a timeout for
waiting for each event, so that it doesn't block indefinately.

Although it watches a specific path, in reality it watches the path's
parent directory, so that it can continue watching the file if it gets
removed and recreated.

1 files changed, 89 insertions, 30 deletions

diff --git a/mullvad-tests/src/lib.rs b/mullvad-tests/src/lib.rs
index 9bc4755827..534bf11003 100644
--- a/mullvad-tests/src/lib.rs
+++ b/mullvad-tests/src/lib.rs
@@ -23,7 +23,7 @@ use std::thread;
 use std::time::{Duration, Instant};
 
 use mullvad_ipc_client::DaemonRpcClient;
-use notify::{op, RawEvent, RecursiveMode, Watcher};
+use notify::{RawEvent, RecommendedWatcher, RecursiveMode, Watcher};
 use openvpn_plugin::types::OpenVpnPluginEvent;
 use os_pipe::{pipe, PipeReader};
 use talpid_ipc::WsIpcClient;
@@ -32,6 +32,8 @@ use tempfile::TempDir;
 use self::mock_openvpn::MOCK_OPENVPN_ARGS_FILE;
 use self::platform_specific::*;
 
+pub use self::notify::op::{self as watch_event, Op as WatchEvent};
+
 type Result<T> = ::std::result::Result<T, String>;
 
 #[cfg(unix)]
@@ -53,44 +55,97 @@ mod platform_specific {
     pub const TALPID_OPENVPN_PLUGIN_FILE: &str = "talpid_openvpn_plugin.dll";
 }
 
-pub fn wait_for_file_write_finish<P: AsRef<Path>>(file_path: P, timeout: Duration) {
-    let file_path = file_path.as_ref();
-    let parent_dir = file_path.parent().expect("Missing file parent directory");
+pub struct PathWatcher {
+    events: mpsc::Receiver<RawEvent>,
+    path: PathBuf,
+    timeout: Duration,
+    _watcher: RecommendedWatcher,
+}
+
+impl PathWatcher {
+    pub fn watch<P: AsRef<Path>>(file_path: P) -> Result<Self> {
+        let file_path = file_path.as_ref();
+        let parent_dir = file_path
+            .parent()
+            .ok_or_else(|| "Missing file parent directory")?;
+
+        let absolute_parent_dir = parent_dir
+            .canonicalize()
+            .map_err(|_| "Failed to get absolute path to watch")?;
+        let file_name = file_path
+            .file_name()
+            .ok_or_else(|| "Missing file name of file path to watch")?;
+        let absolute_file_path = absolute_parent_dir.join(file_name);
 
-    let absolute_parent_dir = parent_dir
-        .canonicalize()
-        .expect("Failed to get absolute path to watch");
-    let file_name = file_path
-        .file_name()
-        .expect("Missing file name of file path to watch");
-    let absolute_file_path = absolute_parent_dir.join(file_name);
+        let (tx, rx) = mpsc::channel();
+        let mut watcher = notify::raw_watcher(tx).map_err(|_| {
+            format!(
+                "Failed to create watcher of file system events to watch {}",
+                file_path.display()
+            )
+        })?;
 
-    let (tx, rx) = mpsc::channel();
-    let mut watcher = notify::raw_watcher(tx).expect("Failed to listen for file system events");
-    let start = Instant::now();
-    let mut remaining_time = Some(timeout);
+        watcher
+            .watch(absolute_parent_dir, RecursiveMode::Recursive)
+            .map_err(|_| {
+                format!(
+                    "Failed to start watching for file system events from {}",
+                    file_path.display()
+                )
+            })?;
+
+        Ok(PathWatcher {
+            events: rx,
+            path: absolute_file_path,
+            timeout: Duration::from_secs(5),
+            _watcher: watcher,
+        })
+    }
 
-    watcher
-        .watch(absolute_parent_dir, RecursiveMode::NonRecursive)
-        .expect("Failed to listen for file system events on directory");
+    pub fn set_timeout(&mut self, timeout: Duration) -> &mut Self {
+        self.timeout = timeout;
+        self
+    }
 
-    if !file_path.exists() {
-        while let Some(wait_time) = remaining_time {
-            let event = rx.recv_timeout(wait_time);
+    pub fn assert_create_write_close_sequence(&mut self) {
+        assert_eq!(self.next(), Some(watch_event::CREATE));
+        assert_eq!(self.next(), Some(watch_event::WRITE));
 
-            if let Ok(RawEvent {
-                path: Some(path),
-                op: Ok(op),
-                ..
-            }) = event
-            {
-                if op.contains(op::CLOSE_WRITE) && path == absolute_file_path {
+        loop {
+            match self.next() {
+                Some(watch_event::WRITE) => continue,
+                event => {
+                    assert_eq!(event, Some(watch_event::CLOSE_WRITE));
                     break;
                 }
             }
+        }
+    }
+}
 
-            remaining_time = timeout.checked_sub(start.elapsed());
+impl Iterator for PathWatcher {
+    type Item = WatchEvent;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let start = Instant::now();
+
+        while let Some(remaining_time) = self.timeout.checked_sub(start.elapsed()) {
+            match self.events.recv_timeout(remaining_time) {
+                Ok(RawEvent {
+                    path: Some(path),
+                    op: Ok(op),
+                    ..
+                }) => if path == self.path {
+                    return Some(op);
+                } else {
+                    continue;
+                },
+                Ok(_) => continue,
+                Err(_) => return None,
+            }
         }
+
+        None
     }
 }
 
@@ -232,7 +287,11 @@ impl DaemonRunner {
 
     pub fn rpc_client(&mut self) -> Result<DaemonRpcClient> {
         if !self.rpc_address_file.exists() {
-            wait_for_file_write_finish(&self.rpc_address_file, Duration::from_secs(10));
+            let _ = PathWatcher::watch(&self.rpc_address_file).map(|mut events| {
+                events
+                    .set_timeout(Duration::from_secs(10))
+                    .find(|&event| event == watch_event::CLOSE_WRITE)
+            });
         }
 
         DaemonRpcClient::with_insecure_rpc_address_file(&self.rpc_address_file)