Merge pull request #4417 from andreaTP/moreRobustServerSpec

fix ServerSpec flaky tests

sbt/src/test/scala/testpkg/ServerSpec.scala

@@ -11,6 +11,7 @@ import org.scalatest._
 import scala.concurrent._
 import scala.annotation.tailrec
 import sbt.protocol.ClientSocket
+import scala.util.Try
 import TestServer.withTestServer
 import java.io.File
 import sbt.io.syntax._
@@ -19,101 +20,129 @@ import sbt.RunFromSourceMain
 import scala.concurrent.ExecutionContext
 import java.util.concurrent.ForkJoinPool
 
-class ServerSpec extends AsyncFreeSpec with Matchers {
+class ServerSpec extends fixture.AsyncFreeSpec with fixture.AsyncTestDataFixture with Matchers {
   "server" - {
-    "should start" in withTestServer("handshake") { p =>
+    "should start" in { implicit td =>
-      p.writeLine(
+      withTestServer("handshake") { p =>
-        """{ "jsonrpc": "2.0", "id": "3", "method": "sbt/setting", "params": { "setting": "root/name" } }"""
+        p.writeLine(
-      )
+          """{ "jsonrpc": "2.0", "id": "3", "method": "sbt/setting", "params": { "setting": "root/name" } }"""
-      assert(p.waitForString(10) { s =>
+        )
-        s contains """"id":"3""""
+        assert(p.waitForString(10) { s =>
-      })
+          s contains """"id":"3""""
+        })
+      }
     }
 
-    "return number id when number id is sent" in withTestServer("handshake") { p =>
+    "return number id when number id is sent" in { implicit td =>
-      p.writeLine(
+      withTestServer("handshake") { p =>
-        """{ "jsonrpc": "2.0", "id": 3, "method": "sbt/setting", "params": { "setting": "root/name" } }"""
+        p.writeLine(
-      )
+          """{ "jsonrpc": "2.0", "id": 3, "method": "sbt/setting", "params": { "setting": "root/name" } }"""
-      assert(p.waitForString(10) { s =>
+        )
-        s contains """"id":3"""
+        assert(p.waitForString(10) { s =>
-      })
+          s contains """"id":3"""
+        })
+      }
     }
 
-    "report task failures in case of exceptions" in withTestServer("events") { p =>
+    "report task failures in case of exceptions" in { implicit td =>
-      p.writeLine(
+      withTestServer("events") { p =>
-        """{ "jsonrpc": "2.0", "id": 11, "method": "sbt/exec", "params": { "commandLine": "hello" } }"""
+        p.writeLine(
-      )
+          """{ "jsonrpc": "2.0", "id": 11, "method": "sbt/exec", "params": { "commandLine": "hello" } }"""
-      assert(p.waitForString(10) { s =>
+        )
-        (s contains """"id":11""") && (s contains """"error":""")
+        assert(p.waitForString(10) { s =>
-      })
+          (s contains """"id":11""") && (s contains """"error":""")
+        })
+      }
     }
 
-    "return error if cancelling non-matched task id" in withTestServer("events") { p =>
+    "return error if cancelling non-matched task id" in { implicit td =>
-      p.writeLine(
+      withTestServer("events") { p =>
-        """{ "jsonrpc": "2.0", "id":12, "method": "sbt/exec", "params": { "commandLine": "run" } }"""
+        p.writeLine(
-      )
+          """{ "jsonrpc": "2.0", "id":12, "method": "sbt/exec", "params": { "commandLine": "run" } }"""
-      p.writeLine(
+        )
-        """{ "jsonrpc": "2.0", "id":13, "method": "sbt/cancelRequest", "params": { "id": "55" } }"""
+        p.writeLine(
-      )
+          """{ "jsonrpc": "2.0", "id":13, "method": "sbt/cancelRequest", "params": { "id": "55" } }"""
+        )
 
-      assert(p.waitForString(20) { s =>
+        assert(p.waitForString(20) { s =>
-        (s contains """"error":{"code":-32800""")
+          (s contains """"error":{"code":-32800""")
-      })
+        })
+      }
     }
 
-    "cancel on-going task with numeric id" in withTestServer("events") { p =>
+    "cancel on-going task with numeric id" in { implicit td =>
-      p.writeLine(
+      withTestServer("events") { p =>
-        """{ "jsonrpc": "2.0", "id":12, "method": "sbt/exec", "params": { "commandLine": "run" } }"""
+        p.writeLine(
-      )
+          """{ "jsonrpc": "2.0", "id":12, "method": "sbt/exec", "params": { "commandLine": "run" } }"""
+        )
 
-      Thread.sleep(1000)
+        assert(p.waitForString(60) { s =>
-
+          p.writeLine(
-      p.writeLine(
+            """{ "jsonrpc": "2.0", "id":13, "method": "sbt/cancelRequest", "params": { "id": "12" } }"""
-        """{ "jsonrpc": "2.0", "id":13, "method": "sbt/cancelRequest", "params": { "id": "12" } }"""
+          )
-      )
+          s contains """"result":{"status":"Task cancelled""""
-
+        })
-      assert(p.waitForString(30) { s =>
+      }
-        s contains """"result":{"status":"Task cancelled""""
-      })
     }
 
-    "cancel on-going task with string id" in withTestServer("events") { p =>
+    "cancel on-going task with string id" in { implicit td =>
-      p.writeLine(
+      withTestServer("events") { p =>
-        """{ "jsonrpc": "2.0", "id": "foo", "method": "sbt/exec", "params": { "commandLine": "run" } }"""
+        p.writeLine(
-      )
+          """{ "jsonrpc": "2.0", "id": "foo", "method": "sbt/exec", "params": { "commandLine": "run" } }"""
+        )
 
-      Thread.sleep(1000)
+        assert(p.waitForString(60) { s =>
-
+          p.writeLine(
-      p.writeLine(
+            """{ "jsonrpc": "2.0", "id": "bar", "method": "sbt/cancelRequest", "params": { "id": "foo" } }"""
-        """{ "jsonrpc": "2.0", "id": "bar", "method": "sbt/cancelRequest", "params": { "id": "foo" } }"""
+          )
-      )
+          s contains """"result":{"status":"Task cancelled""""
-
+        })
-      assert(p.waitForString(30) { s =>
+      }
-        s contains """"result":{"status":"Task cancelled""""
-      })
     }
   }
 }
 
 object TestServer {
-  // The test server instance will be executed in a Thread pool separated from the tests
-  implicit val ec = ExecutionContext.fromExecutor(new ForkJoinPool())
 
   private val serverTestBase: File = new File(".").getAbsoluteFile / "sbt" / "src" / "server-test"
 
-  def withTestServer(testBuild: String)(f: TestServer => Future[Assertion]): Future[Assertion] = {
+  def withTestServer(
+      testBuild: String
+  )(f: TestServer => Future[Assertion])(implicit td: TestData): Future[Assertion] = {
+    println(s"Starting test: ${td.name}")
     IO.withTemporaryDirectory { temp =>
       IO.copyDirectory(serverTestBase / testBuild, temp / testBuild)
-      withTestServer(temp / testBuild)(f)
+      withTestServer(testBuild, temp / testBuild)(f)
     }
   }
 
-  def withTestServer(baseDirectory: File)(f: TestServer => Future[Assertion]): Future[Assertion] = {
+  def withTestServer(testBuild: String, baseDirectory: File)(
-    val testServer = TestServer(baseDirectory)
+      f: TestServer => Future[Assertion]
-    try {
+  )(implicit td: TestData): Future[Assertion] = {
-      f(testServer)
+    // Each test server instance will be executed in a Thread pool separated from the tests
-    } finally {
+    val testServer = TestServer(baseDirectory)(
-      try { testServer.bye() } finally {}
+      ExecutionContext.fromExecutor(new ForkJoinPool())
+    )
+    // checking last log message after initialization
+    // if something goes wrong here the communication streams are corrupted, restarting
+    val init =
+      Try {
+        testServer.waitForString(30) { s =>
+          s contains """"message":"Done""""
+        }
+      }.toOption
+
+    init match {
+      case Some(_) =>
+        try {
+          f(testServer)
+        } finally {
+          try { testServer.bye() } finally {}
+        }
+      case _ =>
+        try { testServer.bye() } finally {}
+        hostLog("Server started but not connected properly... restarting...")
+        withTestServer(testBuild)(f)
     }
   }
 
@@ -216,13 +245,19 @@ case class TestServer(baseDirectory: File)(implicit ec: ExecutionContext) {
 
   @tailrec
   final def waitForString(num: Int)(f: String => Boolean): Boolean = {
-    if (num < 0) false
+    if (num < 0) { throw new Exception("Retries are over.") } else {
-    else
+      // readFrame should be called in another Thread in orrder to be able to time limit it's execution
-      readFrame match {
+      val res = Future { readFrame }(ec)
-        case Some(x) if f(x) => true
+
-        case _ =>
+      import scala.concurrent.duration._
-          waitForString(num - 1)(f)
+      Try {
+        Await.result(res, 1.second)
+      }.toOption.flatten match {
+        // function f should be called in this Thread in order to be executed exactly once before eventually returning
+        case Some(str) if f(str) => true
+        case _                   => waitForString(num - 1)(f)
       }
+    }
   }
 
   def readLine: Option[String] = {