Unit test code with WatchService

Question

Below is a short simple example of using a WatchService to keep data in sync with a file. My question is how to reliably test the code. The test fails occasionally, probably because of a race condition between the os/jvm getting the event into the watch service and the test thread polling the watch service. My desire is to keep the code simple, single threaded, and non blocking but also be testable. I strongly dislike putting sleep calls of arbitrary length into test code. I am hoping there is a better solution.

public class FileWatcher {

private final WatchService watchService;
private final Path path;
private String data;

public FileWatcher(Path path){
    this.path = path;
    try {
        watchService = FileSystems.getDefault().newWatchService();
        path.toAbsolutePath().getParent().register(watchService, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY);
    } catch (Exception ex) {
        throw new RuntimeException(ex);
    }
    load();
}

private void load() {
    try (BufferedReader br = Files.newBufferedReader(path, Charset.defaultCharset())){
        data = br.readLine();
    } catch (IOException ex) {
        data = "";
    }
}

private void update(){
    WatchKey key;
    while ((key=watchService.poll()) != null) {
        for (WatchEvent<?> e : key.pollEvents()) {
            WatchEvent<Path> event = (WatchEvent<Path>) e;
            if (path.equals(event.context())){
                load();
                break;
            }
        }
        key.reset();
    }
}

public String getData(){
    update();
    return data;
}
}

And the current test

public class FileWatcherTest {

public FileWatcherTest() {
}

Path path = Paths.get("myFile.txt");

private void write(String s) throws IOException{
    try (BufferedWriter bw = Files.newBufferedWriter(path, Charset.defaultCharset())) {
        bw.write(s);
    }
}

@Test
public void test() throws IOException{
    for (int i=0; i<100; i++){
        write("hello");
        FileWatcher fw = new FileWatcher(path);
        Assert.assertEquals("hello", fw.getData());
        write("goodbye");
        Assert.assertEquals("goodbye", fw.getData());
    }
}
}

Answer 1

This timing issue is bound to happen because of the polling happening in the watch service.

This test is not really a unit test because it is testing the actual implementation of the default file system watcher.

If I wanted to make a self-contained unit test for this class, I would first modify the FileWatcher so that it does not rely on the default file system. The way I would do this would be to inject a WatchService into the constructor instead of a FileSystem. For example...

public class FileWatcher {

    private final WatchService watchService;
    private final Path path;
    private String data;

    public FileWatcher(WatchService watchService, Path path) {
        this.path = path;
        try {
            this.watchService = watchService;
            path.toAbsolutePath().getParent().register(watchService, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY);
        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }
        load();
    }

    ...

Passing in this dependency instead of the class getting hold of a WatchService by itself makes this class a bit more reusable in the future. For example, what if you wanted to use a different FileSystem implementation (such as an in-memory one like https://github.com/google/jimfs)?

You can now test this class by mocking the dependencies, for example...

import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
import static java.nio.file.StandardWatchEventKinds.ENTRY_DELETE;
import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY;
import static org.fest.assertions.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.nio.file.FileSystem;
import java.nio.file.Path;
import java.nio.file.WatchEvent;
import java.nio.file.WatchKey;
import java.nio.file.WatchService;
import java.nio.file.spi.FileSystemProvider;
import java.util.Arrays;

import org.junit.Before;
import org.junit.Test;

public class FileWatcherTest {

    private FileWatcher fileWatcher;
    private WatchService watchService;

    private Path path;

    @Before
    public void setup() throws Exception {
        // Set up mock watch service and path
        watchService = mock(WatchService.class);

        path = mock(Path.class);

        // Need to also set up mocks for absolute parent path...
        Path absolutePath = mock(Path.class);
        Path parentPath = mock(Path.class);

        // Mock the path's methods...
        when(path.toAbsolutePath()).thenReturn(absolutePath);
        when(absolutePath.getParent()).thenReturn(parentPath);

        // Mock enough of the path so that it can load the test file.
        // On the first load, the loaded data will be "[INITIAL DATA]", any subsequent call it will be "[UPDATED DATA]"
        // (this is probably the smellyest bit of this test...)
        InputStream initialInputStream = createInputStream("[INITIAL DATA]");
        InputStream updatedInputStream = createInputStream("[UPDATED DATA]");
        FileSystem fileSystem = mock(FileSystem.class);
        FileSystemProvider fileSystemProvider = mock(FileSystemProvider.class);

        when(path.getFileSystem()).thenReturn(fileSystem);
        when(fileSystem.provider()).thenReturn(fileSystemProvider);
        when(fileSystemProvider.newInputStream(path)).thenReturn(initialInputStream, updatedInputStream);
        // (end smelly bit)

        // Create the watcher - this should load initial data immediately
        fileWatcher = new FileWatcher(watchService, path);

        // Verify that the watch service was registered with the parent path...
        verify(parentPath).register(watchService, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY);
    }

    @Test
    public void shouldReturnCurrentStateIfNoChanges() {
        // Check to see if the initial data is returned if the watch service returns null on poll...
        when(watchService.poll()).thenReturn(null);
        assertThat(fileWatcher.getData()).isEqualTo("[INITIAL DATA]");
    }

    @Test
    public void shouldLoadNewStateIfFileChanged() {
        // Check that the updated data is loaded when the watch service says the path we are interested in has changed on poll... 
        WatchKey watchKey = mock(WatchKey.class);
        @SuppressWarnings("unchecked")
        WatchEvent<Path> pathChangedEvent = mock(WatchEvent.class);

        when(pathChangedEvent.context()).thenReturn(path);
        when(watchKey.pollEvents()).thenReturn(Arrays.asList(pathChangedEvent));
        when(watchService.poll()).thenReturn(watchKey, (WatchKey) null);

        assertThat(fileWatcher.getData()).isEqualTo("[UPDATED DATA]");
    }

    @Test
    public void shouldKeepCurrentStateIfADifferentPathChanged() {
        // Make sure nothing happens if a different path is updated...
        WatchKey watchKey = mock(WatchKey.class);
        @SuppressWarnings("unchecked")
        WatchEvent<Path> pathChangedEvent = mock(WatchEvent.class);

        when(pathChangedEvent.context()).thenReturn(mock(Path.class));
        when(watchKey.pollEvents()).thenReturn(Arrays.asList(pathChangedEvent));
        when(watchService.poll()).thenReturn(watchKey, (WatchKey) null);

        assertThat(fileWatcher.getData()).isEqualTo("[INITIAL DATA]");
    }

    private InputStream createInputStream(String string) {
        return new ByteArrayInputStream(string.getBytes());
    }

}

I can see why you might want a "real" test for this that does not use mocks - in which case it would not be a unit test and you might not have much choice but to sleep between checks (the JimFS v1.0 code is hard coded to poll every 5 seconds, have not looked at the poll time on the core Java FileSystem's WatchService)

Hope this helps

Answer 2

I created a wrapper around WatchService to clean up many issues I have with the API. It is now much more testable. I am unsure about some of the concurrency issues in PathWatchService though and I have not done thorough testing of it.

New FileWatcher:

public class FileWatcher {

    private final PathWatchService pathWatchService;
    private final Path path;
    private String data;

    public FileWatcher(PathWatchService pathWatchService, Path path) {
        this.path = path;
        this.pathWatchService = pathWatchService;
        try {
            this.pathWatchService.register(path.toAbsolutePath().getParent());
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
        load();
    }

    private void load() {
        try (BufferedReader br = Files.newBufferedReader(path, Charset.defaultCharset())){
            data = br.readLine();
        } catch (IOException ex) {
            data = "";
        }
    }

    public void update(){
        PathEvents pe;
        while ((pe=pathWatchService.poll()) != null) {
            for (WatchEvent we : pe.getEvents()){
                if (path.equals(we.context())){
                    load();
                    return;
                }
            }
        }
    }

    public String getData(){
        update();
        return data;
    }
}

Wrapper:

public class PathWatchService implements AutoCloseable {

    private final WatchService watchService;
    private final BiMap<WatchKey, Path> watchKeyToPath = HashBiMap.create();
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private final Queue<WatchKey> invalidKeys = new ConcurrentLinkedQueue<>();

    /**
     * Constructor.
     */
    public PathWatchService() {
        try {
            watchService = FileSystems.getDefault().newWatchService();
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
    }

    /**
     * Register the input path with the WatchService for all
     * StandardWatchEventKinds. Registering a path which is already being
     * watched has no effect.
     *
     * @param path
     * @return
     * @throws IOException
     */
    public void register(Path path) throws IOException {
        register(path, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY);
    }

    /**
     * Register the input path with the WatchService for the input event kinds.
     * Registering a path which is already being watched has no effect.
     *
     * @param path
     * @param kinds
     * @return
     * @throws IOException
     */
    public void register(Path path, WatchEvent.Kind... kinds) throws IOException {
        try {
            lock.writeLock().lock();
            removeInvalidKeys();
            WatchKey key = watchKeyToPath.inverse().get(path);
            if (key == null) {
                key = path.register(watchService, kinds);
                watchKeyToPath.put(key, path);
            }
        } finally {
            lock.writeLock().unlock();
        }
    }

    /**
     * Close the WatchService.
     *
     * @throws IOException
     */
    @Override
    public void close() throws IOException {
        try {
            lock.writeLock().lock();
            watchService.close();
            watchKeyToPath.clear();
            invalidKeys.clear();
        } finally {
            lock.writeLock().unlock();
        }
    }

    /**
     * Retrieves and removes the next PathEvents object, or returns null if none
     * are present.
     *
     * @return
     */
    public PathEvents poll() {
        return keyToPathEvents(watchService.poll());
    }

    /**
     * Return a PathEvents object from the input key.
     *
     * @param key
     * @return
     */
    private PathEvents keyToPathEvents(WatchKey key) {
        if (key == null) {
            return null;
        }
        try {
            lock.readLock().lock();
            Path watched = watchKeyToPath.get(key);
            List<WatchEvent<Path>> events = new ArrayList<>();
            for (WatchEvent e : key.pollEvents()) {
                events.add((WatchEvent<Path>) e);
            }
            boolean isValid = key.reset();
            if (isValid == false) {
                invalidKeys.add(key);
            }
            return new PathEvents(watched, events, isValid);
        } finally {
            lock.readLock().unlock();
        }
    }

    /**
     * Retrieves and removes the next PathEvents object, waiting if necessary up
     * to the specified wait time, returns null if none are present after the
     * specified wait time.
     *
     * @return
     */
    public PathEvents poll(long timeout, TimeUnit unit) throws InterruptedException {
        return keyToPathEvents(watchService.poll(timeout, unit));
    }

    /**
     * Retrieves and removes the next PathEvents object, waiting if none are yet
     * present.
     *
     * @return
     */
    public PathEvents take() throws InterruptedException {
        return keyToPathEvents(watchService.take());
    }

    /**
     * Get all paths currently being watched. Any paths which were watched but
     * have invalid keys are not returned.
     *
     * @return
     */
    public Set<Path> getWatchedPaths() {
        try {
            lock.readLock().lock();
            Set<Path> paths = new HashSet<>(watchKeyToPath.inverse().keySet());
            WatchKey key;
            while ((key = invalidKeys.poll()) != null) {
                paths.remove(watchKeyToPath.get(key));
            }
            return paths;
        } finally {
            lock.readLock().unlock();
        }
    }

    /**
     * Cancel watching the specified path. Cancelling a path which is not being
     * watched has no effect.
     *
     * @param path
     */
    public void cancel(Path path) {
        try {
            lock.writeLock().lock();
            removeInvalidKeys();
            WatchKey key = watchKeyToPath.inverse().remove(path);
            if (key != null) {
                key.cancel();
            }
        } finally {
            lock.writeLock().unlock();
        }
    }

    /**
     * Removes any invalid keys from internal data structures. Note this
     * operation is also performed during register and cancel calls.
     */
    public void cleanUp() {
        try {
            lock.writeLock().lock();
            removeInvalidKeys();
        } finally {
            lock.writeLock().unlock();
        }
    }

    /**
     * Clean up method to remove invalid keys, must be called from inside an
     * acquired write lock.
     */
    private void removeInvalidKeys() {
        WatchKey key;
        while ((key = invalidKeys.poll()) != null) {
            watchKeyToPath.remove(key);
        }
    }
}

Data class:

public class PathEvents {

    private final Path watched;
    private final ImmutableList<WatchEvent<Path>> events;
    private final boolean isValid;

    /**
     * Constructor.
     * 
     * @param watched
     * @param events
     * @param isValid 
     */
    public PathEvents(Path watched, List<WatchEvent<Path>> events, boolean isValid) {
        this.watched = watched;
        this.events = ImmutableList.copyOf(events);
        this.isValid = isValid;
    }

    /**
     * Return an immutable list of WatchEvent's.
     * @return 
     */
    public List<WatchEvent<Path>> getEvents() {
        return events;
    }

    /**
     * True if the watched path is valid.
     * @return 
     */
    public boolean isIsValid() {
        return isValid;
    }

    /**
     * Return the path being watched in which these events occurred.
     * 
     * @return 
     */
    public Path getWatched() {
        return watched;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        final PathEvents other = (PathEvents) obj;
        if (!Objects.equals(this.watched, other.watched)) {
            return false;
        }
        if (!Objects.equals(this.events, other.events)) {
            return false;
        }
        if (this.isValid != other.isValid) {
            return false;
        }
        return true;
    }

    @Override
    public int hashCode() {
        int hash = 7;
        hash = 71 * hash + Objects.hashCode(this.watched);
        hash = 71 * hash + Objects.hashCode(this.events);
        hash = 71 * hash + (this.isValid ? 1 : 0);
        return hash;
    }

    @Override
    public String toString() {
        return "PathEvents{" + "watched=" + watched + ", events=" + events + ", isValid=" + isValid + '}';
    }
}

And finally the test, note this is not a complete unit test but demonstrates the way to write tests for this situation.

public class FileWatcherTest {

    public FileWatcherTest() {
    }
    Path path = Paths.get("myFile.txt");
    Path parent = path.toAbsolutePath().getParent();

    private void write(String s) throws IOException {
        try (BufferedWriter bw = Files.newBufferedWriter(path, Charset.defaultCharset())) {
            bw.write(s);
        }
    }

    @Test
    public void test() throws IOException, InterruptedException{
        write("hello");

        PathWatchService real = new PathWatchService();
        real.register(parent);
        PathWatchService mock = mock(PathWatchService.class);

        FileWatcher fileWatcher = new FileWatcher(mock, path);
        verify(mock).register(parent);
        Assert.assertEquals("hello", fileWatcher.getData());

        write("goodbye");
        PathEvents pe = real.poll(10, TimeUnit.SECONDS);
        if (pe == null){
            Assert.fail("Should have an event for writing good bye");
        }
        when(mock.poll()).thenReturn(pe).thenReturn(null);

        Assert.assertEquals("goodbye", fileWatcher.getData());
    }
}