Fix test and avoid HTTP server in local mode

Author: sryza

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -1018,6 +1018,7 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
    * supported for Hadoop-supported filesystems.
    */
   def addFile(path: String, recursive: Boolean): Unit = {
+    val isLocalMode = conf.get("spark.master").startsWith("local")
     val uri = new URI(path)
     val schemeCorrectedPath = uri.getScheme match {
       case null | "local" => "file:" + uri.getPath
@@ -1032,18 +1033,20 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
         throw new SparkException(s"Added file $hadoopPath does not exist.")
       }
       val isDir = fs.isDirectory(hadoopPath)
-      if (scheme == "file" && isDir) {
-        throw new SparkException(s"addFile does not support adding local directories.")
+      if (!isLocalMode && scheme == "file" && isDir) {
+        throw new SparkException(s"addFile does not support local directories when not running " +
+          "local mode.")
       }
       if (!recursive && isDir) {
         throw new SparkException(s"Added file $hadoopPath is a directory and recursive is not " +
           "turned on.")
       }
     }
 
-    val key = scheme match {
-      case "file" => env.httpFileServer.addFile(new File(uri.getPath))
-      case _      => path
+    val key = if (!isLocalMode && scheme == "file") {
+      env.httpFileServer.addFile(new File(uri.getPath))
+    } else {
+      schemeCorrectedPath
     }
     val timestamp = System.currentTimeMillis
     addedFiles(key) = timestamp

core/src/test/scala/org/apache/spark/SparkContextSuite.scala

@@ -113,7 +113,18 @@ class SparkContextSuite extends FunSuite with LocalSparkContext {
     }
   }
 
-  test("addFile recursive can't add the same directory twice") {
+  test("addFile recursive can't add directories by default") {
+    val dir = new File("dir")
 
+    try {
+      sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
+      sc.addFile(dir.getAbsolutePath)
+      assert(false, "should have thrown exception")
+    } catch {
+      case _: SparkException =>
+    } finally {
+      sc.stop()
+      dir.delete()
+    }
   }
 }