[BEAM-14510] adding exception tests to LocalFileSystem

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -658,6 +658,7 @@ class BeamModulePlugin implements Plugin<Project> {
         kafka                                       : "org.apache.kafka:kafka_2.11:$kafka_version",
         kafka_clients                               : "org.apache.kafka:kafka-clients:$kafka_version",
         mockito_core                                : "org.mockito:mockito-core:3.7.7",
+        mockito_inline                              : "org.mockito:mockito-inline:4.5.1",
         mongo_java_driver                           : "org.mongodb:mongo-java-driver:3.12.10",
         nemo_compiler_frontend_beam                 : "org.apache.nemo:nemo-compiler-frontend-beam:$nemo_version",
         netty_all                                   : "io.netty:netty-all:$netty_version",

sdks/java/core/build.gradle

@@ -81,6 +81,7 @@ dependencies {
   implementation library.java.antlr_runtime
   implementation library.java.commons_compress
   implementation library.java.commons_lang3
+  testImplementation library.java.mockito_inline
   shadow library.java.jsr305
   shadow library.java.error_prone_annotations
   shadow library.java.jackson_core

sdks/java/core/src/test/java/org/apache/beam/sdk/io/LocalFileSystemTest.java

@@ -26,6 +26,9 @@
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assume.assumeFalse;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import java.io.File;
 import java.io.FileNotFoundException;
@@ -34,6 +37,7 @@
 import java.io.Writer;
 import java.nio.channels.Channels;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Collections;
@@ -57,6 +61,8 @@
 import org.junit.rules.TemporaryFolder;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
 
 /** Tests for {@link LocalFileSystem}. */
 @RunWith(JUnit4.class)
@@ -82,6 +88,29 @@ public void testCreateWithNonExistentSubDirectory() throws Exception {
     testCreate(temporaryFolder.getRoot().toPath().resolve("non-existent-dir").resolve("file.txt"));
   }
 
+  @Test
+  public void testCreateFilesException() throws Exception {
+    File mockParentFile = mock(File.class);
+    when(mockParentFile.mkdirs()).thenThrow(new RuntimeException("exception when creating files"));
+
+    File mockAbsoluteFile = mock(File.class);
+    when(mockAbsoluteFile.getParentFile()).thenReturn(mockParentFile);
+
+    File mockFile = mock(File.class);
+    when(mockFile.getAbsoluteFile()).thenReturn(mockAbsoluteFile);
+
+    Path mockPath = mock(Path.class);
+    when(mockPath.toFile()).thenReturn(mockFile);
+
+    LocalResourceId mockResourceId = mock(LocalResourceId.class);
+    when(mockResourceId.getPath()).thenReturn(mockPath);
+
+    thrown.expect(RuntimeException.class);
+
+    localFileSystem.create(
+        mockResourceId, StandardCreateOptions.builder().setMimeType(MimeTypes.TEXT).build());
+  }
+
   private void testCreate(Path path) throws Exception {
     String expected = "my test string";
     // First with the path string
@@ -152,6 +181,23 @@ public void testCopyWithExistingSrcFile() throws Exception {
         ImmutableList.of(destPath1, destPath2), ImmutableList.of("content1", "content2"));
   }
 
+  @Test
+  public void testCopyWithNonExistingSrcFile() throws Exception {
+    thrown.expect(NoSuchFileException.class);
+
+    Path existentSrc = temporaryFolder.newFile().toPath();
+    Path nonExistentSrc = temporaryFolder.getRoot().toPath().resolve("non-existent-file.txt");
+
+    Path destPath1 = temporaryFolder.getRoot().toPath().resolve("nonexistentdir").resolve("dest1");
+    Path destPath2 = destPath1.resolveSibling("dest2");
+
+    createFileWithContent(existentSrc, "content");
+
+    localFileSystem.copy(
+        toLocalResourceIds(ImmutableList.of(existentSrc, nonExistentSrc), false /* isDirectory */),
+        toLocalResourceIds(ImmutableList.of(destPath1, destPath2), false /* isDirectory */));
+  }
+
   @Test
   public void testMoveWithExistingSrcFile() throws Exception {
     Path srcPath1 = temporaryFolder.newFile().toPath();
@@ -174,6 +220,50 @@ public void testMoveWithExistingSrcFile() throws Exception {
     assertFalse(srcPath2 + "exists", srcPath2.toFile().exists());
   }
 
+  @Test
+  public void testMoveWithNonExistingSrcFile() throws Exception {
+    Path existingSrc = temporaryFolder.newFile().toPath();
+    Path nonExistingSrc = temporaryFolder.getRoot().toPath().resolve("non-existent-file.txt");
+
+    Path destPath1 = temporaryFolder.getRoot().toPath().resolve("nonexistentdir").resolve("dest1");
+    Path destPath2 = destPath1.resolveSibling("dest2");
+
+    createFileWithContent(existingSrc, "content1");
+
+    thrown.expect(NoSuchFileException.class);
+
+    localFileSystem.rename(
+        toLocalResourceIds(ImmutableList.of(existingSrc, nonExistingSrc), false /* isDirectory */),
+        toLocalResourceIds(ImmutableList.of(destPath1, destPath2), false /* isDirectory */));
+  }
+
+  @Test
+  public void testMoveFilesWithException() throws Exception {
+    Path srcPath1 = temporaryFolder.newFile().toPath();
+    Path srcPath2 = temporaryFolder.newFile().toPath();
+
+    Path destPath1 = temporaryFolder.getRoot().toPath().resolve("nonexistentdir").resolve("dest1");
+    Path destPath2 = srcPath2.resolveSibling("dest2");
+
+    createFileWithContent(srcPath1, "content1");
+    createFileWithContent(srcPath2, "content2");
+
+    try (MockedStatic<java.nio.file.Files> mockFiles =
+        Mockito.mockStatic(java.nio.file.Files.class)) {
+      System.out.println(srcPath1 + " plus " + destPath1);
+
+      mockFiles
+          .when(() -> java.nio.file.Files.move(any(), any(), any(), any()))
+          .thenThrow(new RuntimeException("exception while moving files"));
+
+      thrown.expect(RuntimeException.class);
+
+      localFileSystem.rename(
+          toLocalResourceIds(ImmutableList.of(srcPath1, srcPath2), false /* isDirectory */),
+          toLocalResourceIds(ImmutableList.of(destPath1, destPath2), false /* isDirectory */));
+    }
+  }
+
   @Test
   public void testDelete() throws Exception {
     File f1 = temporaryFolder.newFile("file1");
@@ -446,6 +536,23 @@ public void testMatchNewResource() {
     assertTrue(exception.getMessage().startsWith("Expected file path but received directory path"));
   }
 
+  @Test
+  public void testGetDefaultFileSystemException() throws Exception {
+    temporaryFolder.newFile("a");
+
+    try (MockedStatic<java.nio.file.FileSystems> mockFileSystems =
+        Mockito.mockStatic(java.nio.file.FileSystems.class)) {
+      mockFileSystems
+          .when(java.nio.file.FileSystems::getDefault)
+          .thenThrow(new RuntimeException("exception finding filesystem"));
+
+      thrown.expect(RuntimeException.class);
+
+      localFileSystem.match(
+          ImmutableList.of(temporaryFolder.getRoot().toPath().resolve("a").toString()));
+    }
+  }
+
   private void createFileWithContent(Path path, String content) throws Exception {
     try (Writer writer =
         Channels.newWriter(