Merge pull request scala#8087 from som-snytt/topic/test-cleanup

Suppress stack trace in test

Author: lrytz

src/testkit/scala/tools/testkit/AssertUtil.scala

@@ -10,18 +10,19 @@
  * additional information regarding copyright ownership.
  */
 
- package scala.tools.testkit
+package scala.tools.testkit
 
 import org.junit.Assert, Assert._
 import scala.reflect.ClassTag
 import scala.runtime.ScalaRunTime.stringOf
-import scala.collection.GenIterable
 import scala.collection.JavaConverters._
 import scala.collection.mutable
-import scala.concurrent.{Await, Awaitable, SyncVar, TimeoutException}
-import scala.util.Try
+import scala.concurrent.{Await, Awaitable}
+import scala.util.{Failure, Success, Try}
 import scala.util.Properties.isJavaAtLeast
 import scala.util.control.NonFatal
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+import java.util.concurrent.atomic.AtomicReference
 import java.lang.ref._
 import java.lang.reflect.{Array => _, _}
 import java.util.IdentityHashMap
@@ -129,14 +130,18 @@ object AssertUtil {
 
   /** Assert no new threads, with some margin for arbitrary threads to exit. */
   def assertZeroNetThreads(body: => Unit): Unit = {
-    val result = new SyncVar[Option[Throwable]]
     val group = new ThreadGroup("junit")
-    def check() = {
+    try assertZeroNetThreads(group)(body)
+    finally group.destroy()
+  }
+  def assertZeroNetThreads[A](group: ThreadGroup)(body: => A): Try[A] = {
+    val testDone = new CountDownLatch(1)
+    def check(): Try[A] = {
       val beforeCount = group.activeCount
       val beforeThreads = new Array[Thread](beforeCount)
       assertEquals("Spurious early thread creation.", beforeCount, group.enumerate(beforeThreads))
 
-      body
+      val outcome = Try(body)
 
       val afterCount = {
         waitForIt(group.activeCount <= beforeCount, label = "after count")
@@ -146,32 +151,47 @@ object AssertUtil {
       assertEquals("Spurious late thread creation.", afterCount, group.enumerate(afterThreads))
       val staleThreads = afterThreads.toList.diff(beforeThreads)
       //staleThreads.headOption.foreach(_.getStackTrace.foreach(println))
-      assertEquals(staleThreads.mkString("There are stale threads: ",",",""), beforeCount, afterCount)
-      assertTrue(staleThreads.mkString("There are stale threads: ",",",""), staleThreads.isEmpty)
+      val staleMessage = staleThreads.mkString("There are stale threads: ",",","")
+      assertEquals(staleMessage, beforeCount, afterCount)
+      assertTrue(staleMessage, staleThreads.isEmpty)
+
+      outcome
     }
-    def test() = {
+    val result = new AtomicReference[Try[A]]()
+    def test(): Try[A] =
       try {
-        check()
-        result.put(None)
-      } catch {
-        case t: Throwable => result.put(Some(t))
+        val checked = check()
+        result.set(checked)
+        checked
+      } finally {
+        testDone.countDown()
       }
-    }
-    val timeout = 10 * 1000L  // last chance timeout
+
+    val timeout = 10 * 1000L
     val thread = new Thread(group, () => test())
-    def resulted: Boolean = result.get(timeout).isDefined
+    def abort(): Try[A] = {
+      group.interrupt()
+      new Failure(new AssertionError("Test did not complete"))
+    }
     try {
       thread.start()
-      waitForIt(resulted, Slow, label = "test result")
-      val err = result.take(timeout)
-      err.foreach(e => throw e)
+      waitForIt(testDone.getCount == 0, Fast, label = "test result")
+      if (testDone.await(timeout, TimeUnit.MILLISECONDS))
+        result.get
+      else
+        abort()
     } finally {
       thread.join(timeout)
-      group.destroy()
     }
   }
 
   /** Wait for a condition, with a simple back-off strategy.
+   *
+   *  This makes it easier to see hanging threads in development
+   *  without tweaking a timeout parameter. Conversely, when a thread
+   *  fails to make progress in a test environment, we allow the wait
+   *  period to grow larger than usual, since a long wait for failure
+   *  is acceptable.
    *
    *  It would be nicer if what we're waiting for gave us
    *  a progress indicator: we don't care if something
@@ -213,9 +233,51 @@ object AssertUtil {
 
   /** Like Await.ready but return false on timeout, true on completion, throw InterruptedException. */
   def readyOrNot(awaitable: Awaitable[_]): Boolean = Try(Await.ready(awaitable, TestDuration.Standard)).isSuccess
+
+  def withoutATrace[A](body: => A) = NoTrace(body)
 }
 
 object TestDuration {
   import scala.concurrent.duration.{Duration, SECONDS}
   val Standard = Duration(4, SECONDS)
 }
+
+/** Run a thunk, collecting uncaught exceptions from any spawned threads. */
+class NoTrace[A](body: => A) extends Runnable {
+
+  private val uncaught = new mutable.ListBuffer[(Thread, Throwable)]()
+
+  @volatile private[testkit] var result: Option[A] = None
+
+  def run(): Unit = {
+    import AssertUtil.assertZeroNetThreads
+    val group = new ThreadGroup("notrace") {
+      override def uncaughtException(t: Thread, e: Throwable): Unit = synchronized {
+        uncaught += ((t, e))
+      }
+    }
+    try assertZeroNetThreads(group)(body) match {
+      case Success(a) => result = Some(a)
+      case Failure(e) => synchronized { uncaught += ((Thread.currentThread, e)) }
+    }
+    finally group.destroy()
+  }
+
+  private[testkit] lazy val errors: List[(Thread, Throwable)] = synchronized(uncaught.toList)
+
+  private def suppress(t: Throwable, other: Throwable): t.type = { t.addSuppressed(other) ; t }
+
+  private final val noError = None: Option[Throwable]
+
+  def asserted: Option[Throwable] = 
+    errors.collect { case (_, e: AssertionError) => e }
+      .foldLeft(noError)((res, e) => res.map(suppress(_, e)).orElse(Some(e)))
+
+  def apply(test: (Option[A], List[(Thread, Throwable)]) => Option[Throwable]) = {
+    run()
+    test(result, errors).orElse(asserted).foreach(e => throw e)
+  }
+}
+object NoTrace {
+  def apply[A](body: => A): NoTrace[A] = new NoTrace(body)
+}

test/junit/scala/sys/process/ProcessTest.scala

@@ -1,6 +1,6 @@
 package scala.sys.process
 
-import java.io.{ByteArrayInputStream, File}
+import java.io.{ByteArrayInputStream, File, IOException}
 import java.nio.file.{Files, Paths}, Files.createTempFile
 import java.nio.charset.StandardCharsets.UTF_8
 
@@ -10,6 +10,7 @@ import scala.util.Try
 //import scala.sys.process._
 import scala.util.Properties._
 import scala.collection.JavaConverters._
+import scala.tools.testkit.AssertUtil._
 
 import org.junit.runner.RunWith
 import org.junit.runners.JUnit4
@@ -89,7 +90,7 @@ class ProcessTest {
       Files.write(file, List(prefix).asJava, UTF_8)
       file
     }
-    val file1 = Paths.get("total", "junk")
+    val noFile = Paths.get("total", "junk")
     val p2 = new ProcessMock(false)
     val failed = new java.util.concurrent.atomic.AtomicBoolean
     val pb2 = new ProcessBuilderMock(p2, error = true) {
@@ -102,14 +103,23 @@ class ProcessTest {
     val out = createTempFile("out", "tmp")
     val outf = out.toFile
 
-    try {
-      val p0 = (file1.toFile : ProcessBuilder.Source).cat #&& pb2
-      val p = p0 #> outf
+    def process =
+      try {
+        val p0 = (noFile.toFile : ProcessBuilder.Source).cat #&& pb2
+        val p = p0 #> outf
 
-      assertEquals(1, p.!)
-      assertFalse(failed.get)
-    } finally {
-      Files.delete(out)
+        assertEquals(1, p.!)
+        assertFalse(failed.get)
+      } finally {
+        Files.delete(out)
+      }
+
+    def fail(why: String): Option[Throwable] = Some(new AssertionError(why))
+
+    withoutATrace(process) {
+      case (None, _) => fail("No main result")
+      case (_, (_, (_: IOException)) :: Nil) => None
+      case (_, other) => fail(s"Expected one IOException, got $other")
     }
   }
 }

test/junit/scala/tools/testing/AssertUtilTest.scala

@@ -17,4 +17,64 @@ class AssertUtilTest {
     val r = new SoftReference(o)
     assertNotReachable(o, new Holder(r)) { }
   }
+
+  @Test def `asserts on child threads are suppressed`(): Unit = {
+    def kickoff(body: => Unit): Unit = {
+      val t = new Thread(() => body)
+      t.start()
+      t.join()
+    }
+    val sut = withoutATrace {
+      kickoff(assertEquals(42, 17))
+      kickoff(???)
+      kickoff(assertEquals("hi", "bi"))
+      kickoff(assertEquals("low", "brow"))
+    }
+    sut.run()
+    sut.asserted match {
+      case None    => fail("Expected assertion errors")
+      case Some(e) => assertEquals(2, e.getSuppressed.length)
+    }
+  }
+
+  @Test def `waits for child threads to complete`(): Unit = {
+    import java.util.concurrent.CountDownLatch
+    val latch = new CountDownLatch(1)
+    def kickoff(body: => Unit): Unit = {
+      val t = new Thread(() => body)
+      t.start()
+    }
+    val sut = withoutATrace {
+      kickoff {
+        latch.await()
+        assertEquals(42, 17)  // must wait to see this
+      }
+      kickoff {
+        Thread.sleep(100L)    // make them wait for it
+        latch.countDown()
+      }
+      kickoff(assertEquals("hi", "bi"))    // ordinary background thread assertion
+      assertEquals("low", "brow")          // "foreground" thread assertion must be handled
+    }
+    sut.run()
+    sut.asserted match {
+      case None    => fail("Expected assertion errors")
+      case Some(e) => assertEquals(2, e.getSuppressed.length)
+    }
+  }
+
+  @Test def `result is returned`(): Unit = {
+    def kickoff(body: => Unit): Unit = new Thread(() => body).start()
+    def f() = {
+      kickoff {
+        assertEquals(42, 17)
+      }
+      27
+    }
+    val sut = withoutATrace(f())
+    sut.run()
+    assertEquals(Some(27), sut.result)
+    assertEquals(1, sut.errors.size)
+    assertEquals(0, sut.errors.head._2.getSuppressed.length)
+  }
 }