You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@toree.apache.org by ch...@apache.org on 2016/04/22 16:10:20 UTC
[26/32] incubator-toree git commit: Removed system tests
Removed system tests
Project: http://git-wip-us.apache.org/repos/asf/incubator-toree/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-toree/commit/a7f50032
Tree: http://git-wip-us.apache.org/repos/asf/incubator-toree/tree/a7f50032
Diff: http://git-wip-us.apache.org/repos/asf/incubator-toree/diff/a7f50032
Branch: refs/heads/master
Commit: a7f500324e1ca5a6862cb2c6ebebe169a0b30a19
Parents: 23c3abb
Author: Corey A. Stubbs <cs...@us.ibm.com>
Authored: Mon Apr 11 09:29:34 2016 -0500
Committer: Gino Bustelo <lb...@apache.org>
Committed: Thu Apr 21 14:47:41 2016 -0500
----------------------------------------------------------------------
.../apache/toree/utils/TaskManagerSpec.scala | 2 +-
.../scala/system/KernelCommSpecForSystem.scala | 456 -------------------
.../src/test/scala/system/SuiteForSystem.scala | 38 --
.../src/test/scala/system/TruncationTests.scala | 228 ----------
.../test/utils/NoArgSparkKernelTestKit.scala | 27 --
.../scala/test/utils/SparkKernelDeployer.scala | 186 --------
6 files changed, 1 insertion(+), 936 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/a7f50032/kernel-api/src/test/scala/org/apache/toree/utils/TaskManagerSpec.scala
----------------------------------------------------------------------
diff --git a/kernel-api/src/test/scala/org/apache/toree/utils/TaskManagerSpec.scala b/kernel-api/src/test/scala/org/apache/toree/utils/TaskManagerSpec.scala
index 3f97cbb..61ae715 100644
--- a/kernel-api/src/test/scala/org/apache/toree/utils/TaskManagerSpec.scala
+++ b/kernel-api/src/test/scala/org/apache/toree/utils/TaskManagerSpec.scala
@@ -34,7 +34,7 @@ class TaskManagerSpec extends FunSpec with Matchers with MockitoSugar
with Eventually with Timeouts
{
implicit override val patienceConfig = PatienceConfig(
- timeout = scaled(Span(200, Milliseconds)),
+ timeout = scaled(Span(30, Seconds)),
interval = scaled(Span(5, Milliseconds))
)
private val MaxTestTasks = 50000
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/a7f50032/kernel/src/test/scala/system/KernelCommSpecForSystem.scala
----------------------------------------------------------------------
diff --git a/kernel/src/test/scala/system/KernelCommSpecForSystem.scala b/kernel/src/test/scala/system/KernelCommSpecForSystem.scala
deleted file mode 100644
index 9ebe146..0000000
--- a/kernel/src/test/scala/system/KernelCommSpecForSystem.scala
+++ /dev/null
@@ -1,456 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License
- */
-
-package system
-
-import akka.testkit.TestProbe
-import org.apache.toree.communication.ZMQMessage
-import org.apache.toree.kernel.protocol.v5
-import org.apache.toree.kernel.protocol.v5.content._
-import org.apache.toree.kernel.protocol.v5.kernel.ActorLoader
-import org.apache.toree.kernel.protocol.v5.kernel.Utilities._
-import org.apache.toree.kernel.protocol.v5.{KMBuilder, KernelMessage, SocketType}
-import org.scalatest._
-import play.api.libs.json.Json
-import test.utils.NoArgSparkKernelTestKit
-
-import scala.concurrent.duration._
-
-
-/**
- * Used for system-wide (entire Spark Kernel) tests relating to Comm usage.
- *
- * @note Do not use non-generated target names or Comm ids! This is a shared
- * kernel instance (to avoid death by slowness) and it is not guaranteed
- * that previous tests will do proper cleanup!
- */
-@DoNotDiscover
-class KernelCommSpecForSystem
- extends NoArgSparkKernelTestKit
- with FunSpecLike with Matchers
-{
- private val MaxFishTime = 5.seconds
-
- import test.utils.SparkKernelDeployer._
-
- private def waitForExecuteReply(
- shell: TestProbe, headerId: v5.UUID, maxTime: Duration = MaxFishTime
- ): Unit =
- shell.fishForMessage(maxTime) {
- case KernelMessage(_, _, header, parentHeader, _, _)
- if header.msg_type == ExecuteReply.toTypeString &&
- parentHeader.msg_id == headerId=> true
- case _ => false
- }
-
- private def waitForBusyStatus(
- ioPub: TestProbe, headerId: v5.UUID, maxTime: Duration = MaxFishTime
- ): Unit =
- ioPub.fishForMessage(maxTime) {
- case KernelMessage(_, _, header, parentHeader, _, contentString)
- if header.msg_type == KernelStatus.toTypeString &&
- parentHeader.msg_id == headerId =>
- Json.parse(contentString).as[KernelStatus].execution_state == "busy"
- case _ => false
- }
-
- private def waitForIdleStatus(
- ioPub: TestProbe, headerId: v5.UUID, maxTime: Duration = MaxFishTime
- ): Unit =
- ioPub.fishForMessage(maxTime) {
- case KernelMessage(_, _, header, parentHeader, _, contentString)
- if header.msg_type == KernelStatus.toTypeString &&
- parentHeader.msg_id == headerId =>
- Json.parse(contentString).as[KernelStatus].execution_state == "idle"
- case _ => false
- }
-
- private def setupCallback(
- actorLoader: ActorLoader, shell: TestProbe, callbackFunc: String,
- targetName: String, expectedMessage: String
- ) = {
- // If open callback, there are four arguments
- // If msg/close callback, there are three arguments
- val callbackArgs =
- if (callbackFunc.toLowerCase.trim == "addopenhandler") """(_,_,_,_)"""
- else """(_,_,_)"""
-
- // Set up a callback (using stream messages as indicator)
- val codeRequest = s"""
- |kernel.comm.register("$targetName").$callbackFunc {
- | $callbackArgs => kernel.out.println("$expectedMessage")
- |}
- """.stripMargin.trim
- val (messageId, message) = buildZMQExecuteRequest(codeRequest)
- actorLoader.load(SocketType.Shell) ! message
-
- waitForExecuteReply(shell, messageId)
- }
-
- private def testForCallback(
- ioPub: TestProbe, expectedMessage: String, maxTime: Duration = MaxFishTime
- ) = {
- // Should discover an outgoing kernel message for stream
- ioPub.fishForMessage(maxTime) {
- case KernelMessage(_, _, header, _, _, contentString)
- if header.msg_type == StreamContent.toTypeString =>
- val callbackMessage =
- Json.parse(contentString).as[StreamContent].text.trim
- callbackMessage == expectedMessage
- case _ => false
- }
- }
-
- private def buildZMQExecuteRequest(codeRequest: String): (v5.UUID, ZMQMessage) = {
- val kernelMessage = KMBuilder()
- .withHeader(ExecuteRequest.toTypeString)
- .withContentString(ExecuteRequest(
- code = codeRequest,
- silent = false,
- store_history = false,
- user_expressions = v5.UserExpressions(),
- allow_stdin = true
- )).build
-
- (kernelMessage.header.msg_id, kernelMessage)
- }
-
- // Not using TestCommId to avoid problems if previous comm using that
- // id in tests was not properly closed
- private def buildZMQCommOpen(
- targetName: String, data: v5.MsgData,
- commId: v5.UUID = java.util.UUID.randomUUID().toString
- ): (v5.UUID, ZMQMessage) = {
- val kernelMessage = KMBuilder()
- .withHeader(CommOpen.toTypeString)
- .withContentString(CommOpen(
- comm_id = commId,
- target_name = targetName,
- data = data
- )).build
-
- (kernelMessage.header.msg_id, kernelMessage)
- }
-
- // Not using TestCommId to avoid problems if previous comm using that
- // id in tests was not properly closed
- private def buildZMQCommMsg(commId: v5.UUID, data: v5.MsgData): (v5.UUID, ZMQMessage) = {
- val kernelMessage = KMBuilder()
- .withHeader(CommMsg.toTypeString)
- .withContentString(CommMsg(
- comm_id = commId,
- data = data
- )).build
-
- (kernelMessage.header.msg_id, kernelMessage)
- }
-
- // Not using TestCommId to avoid problems if previous comm using that
- // id in tests was not properly closed
- private def buildZMQCommClose(commId: v5.UUID, data: v5.MsgData): (v5.UUID, ZMQMessage) = {
- val kernelMessage = KMBuilder()
- .withHeader(CommClose.toTypeString)
- .withContentString(CommClose(
- comm_id = commId,
- data = data
- )).build
-
- (kernelMessage.header.msg_id, kernelMessage)
- }
-
- describe("Comm for Kernel System") {
- describe("executing Comm API to open a new comm") {
- it("should correctly send comm_open to socket") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- val testTargetName = java.util.UUID.randomUUID().toString
- val codeRequest =
- s"""
- |kernel.comm.open("$testTargetName")
- """.stripMargin.trim
-
- // Send our code to trigger the Comm API
- val (_, message) = buildZMQExecuteRequest(codeRequest)
- actorLoader.load(SocketType.Shell) ! message
-
- // Should discover an outgoing kernel message for comm_open
- ioPub.fishForMessage(MaxFishTime) {
- case KernelMessage(_, _, header, _, _, _)
- if header.msg_type == CommOpen.toTypeString => true
- case _ => false
- }
- }
- }
- }
-
- describe("executing Comm API to send a message") {
- it("should correctly send comm_msg to socket") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- val testTargetName = java.util.UUID.randomUUID().toString
- val codeRequest =
- s"""
- |val commWriter = kernel.comm.open("$testTargetName")
- |commWriter.writeMsg(
- | org.apache.toree.kernel.protocol.v5.MsgData("key" -> "value")
- |)
- """.stripMargin.trim
-
- // Send our code to trigger the Comm API
- val (_, message) = buildZMQExecuteRequest(codeRequest)
- actorLoader.load(SocketType.Shell) ! message
-
- // Should discover an outgoing kernel message for comm_msg
- ioPub.fishForMessage(MaxFishTime) {
- case KernelMessage(_, _, header, _, _, contentString)
- if header.msg_type == CommMsg.toTypeString =>
- (Json.parse(contentString).as[CommMsg].data \ "key").as[String] == "value"
- case _ => false
- }
- }
- }
- }
-
- describe("executing Comm API to close an existing comm") {
- it("should correctly send comm_close to socket") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- val testTargetName = java.util.UUID.randomUUID().toString
- val codeRequest =
- s"""
- |val commWriter = kernel.comm.open("$testTargetName")
- |commWriter.close()
- """.stripMargin.trim
-
- // Send our code to trigger the Comm API
- val (_, message) = buildZMQExecuteRequest(codeRequest)
- actorLoader.load(SocketType.Shell) ! message
-
- // Should discover an outgoing kernel message for comm_close
- ioPub.fishForMessage(MaxFishTime) {
- case KernelMessage(_, _, header, _, _, _)
- if header.msg_type == CommClose.toTypeString => true
- case _ => false
- }
- }
- }
- }
-
- describe("receiving Comm API open from a client") {
- it("should respond comm_close if the target is not found") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- val testTargetName = java.util.UUID.randomUUID().toString
-
- // Send a comm_open (as if a client did it)
- val (_, message) = buildZMQCommOpen(testTargetName, v5.MsgData.Empty)
- actorLoader.load(SocketType.Shell) ! message
-
- // Should discover an outgoing kernel message for comm_close
- ioPub.fishForMessage(MaxFishTime) {
- case KernelMessage(_, _, header, _, _, _)
- if header.msg_type == CommClose.toTypeString => true
- case _ => false
- }
- }
- }
-
- it("should not execute open callbacks if the target is not found") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- val testTargetName = java.util.UUID.randomUUID().toString
- val expectedMessage = "callback" + testTargetName
-
- setupCallback(actorLoader, shell, "addOpenHandler",
- testTargetName, expectedMessage)
-
- // Send a comm_open (as if a client did it) with wrong target
- val (_, message) =
- buildZMQCommOpen(testTargetName + "wrong", v5.MsgData.Empty)
- actorLoader.load(SocketType.Shell) ! message
-
- // Throws an assertion error when a timeout occurs
- intercept[AssertionError] {
- testForCallback(ioPub, expectedMessage, 3.seconds)
- }
- }
- }
-
- it("should execute open callbacks if the target is found") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- val testTargetName = java.util.UUID.randomUUID().toString
- val expectedMessage = "callback" + testTargetName
-
- setupCallback(actorLoader, shell, "addOpenHandler",
- testTargetName, expectedMessage)
-
- // Send a comm_open (as if a client did it)
- val (_, message) = buildZMQCommOpen(testTargetName, v5.MsgData.Empty)
- actorLoader.load(SocketType.Shell) ! message
-
- testForCallback(ioPub, expectedMessage)
- }
- }
-
- // Tested through other tests
- /*ignore("should link the target and Comm id if the target is found") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- }
- }*/
- }
-
- describe("receiving Comm API message from a client") {
- it("should not execute message callbacks if the Comm id is not found") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- val testTargetName = java.util.UUID.randomUUID().toString
- val testCommId = java.util.UUID.randomUUID().toString
- val expectedMessage = "callback" + testTargetName
-
- setupCallback(actorLoader, shell, "addMsgHandler",
- testTargetName, expectedMessage)
-
- // Set up the Comm id prior so we have a proper link
- val (openId, openMessage) =
- buildZMQCommOpen(testTargetName, v5.MsgData.Empty, testCommId)
- actorLoader.load(SocketType.Shell) ! openMessage
-
- waitForIdleStatus(ioPub, openId)
-
- // Send a comm_msg (as if a client did it) with wrong id
- val (msgId, msgMessage) =
- buildZMQCommMsg(testCommId + "wrong", v5.MsgData.Empty)
- actorLoader.load(SocketType.Shell) ! msgMessage
-
- // Throws an assertion error when a timeout occurs
- intercept[AssertionError] {
- testForCallback(ioPub, expectedMessage, 3.seconds)
- }
- }
- }
-
- it("should execute message callbacks if the Comm id is found") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- val testTargetName = java.util.UUID.randomUUID().toString
- val testCommId = java.util.UUID.randomUUID().toString
- val expectedMessage = "callback" + testTargetName
-
- setupCallback(actorLoader, shell, "addMsgHandler",
- testTargetName, expectedMessage)
-
- // Set up the Comm id prior so we have a proper link
- val (openId, openMessage) =
- buildZMQCommOpen(testTargetName, v5.MsgData.Empty, testCommId)
- actorLoader.load(SocketType.Shell) ! openMessage
-
- waitForIdleStatus(ioPub, openId)
-
- // Send a comm_msg (as if a client did it)
- val (msgId, msgMessage) =
- buildZMQCommMsg(testCommId, v5.MsgData.Empty)
- actorLoader.load(SocketType.Shell) ! msgMessage
-
- testForCallback(ioPub, expectedMessage)
- }
- }
- }
-
- describe("receiving Comm API close from a client") {
- it("should not execute close callbacks if the Comm id is not found") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- val testTargetName = java.util.UUID.randomUUID().toString
- val testCommId = java.util.UUID.randomUUID().toString
- val expectedMessage = "callback" + testTargetName
-
- setupCallback(actorLoader, shell, "addCloseHandler",
- testTargetName, expectedMessage)
-
- // Set up the Comm id prior so we have a proper link
- val (openId, openMessage) =
- buildZMQCommOpen(testTargetName, v5.MsgData.Empty, testCommId)
- actorLoader.load(SocketType.Shell) ! openMessage
-
- waitForIdleStatus(ioPub, openId)
-
- // Send a comm_close (as if a client did it) with wrong id
- val (closeId, closeMessage) =
- buildZMQCommClose(testCommId + "wrong", v5.MsgData.Empty)
- actorLoader.load(SocketType.Shell) ! closeMessage
-
- // Throws an assertion error when a timeout occurs
- intercept[AssertionError] {
- testForCallback(ioPub, expectedMessage, 3.seconds)
- }
- }
- }
-
- it("should execute close callbacks if the Comm id is found") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- val testTargetName = java.util.UUID.randomUUID().toString
- val testCommId = java.util.UUID.randomUUID().toString
- val expectedMessage = "callback" + testTargetName
-
- setupCallback(actorLoader, shell, "addCloseHandler",
- testTargetName, expectedMessage)
-
- // Set up the Comm id prior so we have a proper link
- val (openId, openMessage) =
- buildZMQCommOpen(testTargetName, v5.MsgData.Empty, testCommId)
- actorLoader.load(SocketType.Shell) ! openMessage
-
- waitForIdleStatus(ioPub, openId)
-
- // Send a comm_close (as if a client did it)
- val (closeId, closeMessage) =
- buildZMQCommClose(testCommId, v5.MsgData.Empty)
- actorLoader.load(SocketType.Shell) ! closeMessage
-
- testForCallback(ioPub, expectedMessage)
- }
- }
-
- it("should unlink the Comm id from the target if the Comm id is found") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- val testTargetName = java.util.UUID.randomUUID().toString
- val testCommId = java.util.UUID.randomUUID().toString
- val expectedMessage = "callback" + testTargetName
-
- setupCallback(actorLoader, shell, "addCloseHandler",
- testTargetName, expectedMessage)
-
- // Set up the Comm id prior so we have a proper link
- val (openId, openMessage) =
- buildZMQCommOpen(testTargetName, v5.MsgData.Empty, testCommId)
- actorLoader.load(SocketType.Shell) ! openMessage
-
- waitForIdleStatus(ioPub, openId)
-
- // Send a comm_close (as if a client did it)
- val (closeId, closeMessage) =
- buildZMQCommClose(testCommId, v5.MsgData.Empty)
- actorLoader.load(SocketType.Shell) ! closeMessage
-
- testForCallback(ioPub, expectedMessage)
-
- // Send a comm_close (again) with the expectation that nothing happens
- val (closeId2, closeMessage2) =
- buildZMQCommClose(testCommId, v5.MsgData.Empty)
- actorLoader.load(SocketType.Shell) ! closeMessage2
-
- intercept[AssertionError] {
- testForCallback(ioPub, expectedMessage)
- }
- }
- }
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/a7f50032/kernel/src/test/scala/system/SuiteForSystem.scala
----------------------------------------------------------------------
diff --git a/kernel/src/test/scala/system/SuiteForSystem.scala b/kernel/src/test/scala/system/SuiteForSystem.scala
deleted file mode 100644
index 4556e5a..0000000
--- a/kernel/src/test/scala/system/SuiteForSystem.scala
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License
- */
-
-package system
-
-import org.scalatest.{BeforeAndAfterAll, Suites}
-import test.utils.SparkKernelDeployer
-
-class SuiteForSystem extends Suites(
- new KernelCommSpecForSystem,
- new TruncationTests
-) with BeforeAndAfterAll {
- override protected def beforeAll(): Unit = {
- // Initialize the kernel for system tests
- println("Initializing kernel for system tests")
- SparkKernelDeployer.noArgKernelBootstrap
- }
-
- override protected def afterAll(): Unit = {
- println("Shutting down kernel for system tests")
- SparkKernelDeployer.noArgKernelBootstrap.shutdown()
- println("Finished shutting down kernel!")
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/a7f50032/kernel/src/test/scala/system/TruncationTests.scala
----------------------------------------------------------------------
diff --git a/kernel/src/test/scala/system/TruncationTests.scala b/kernel/src/test/scala/system/TruncationTests.scala
deleted file mode 100644
index 118a609..0000000
--- a/kernel/src/test/scala/system/TruncationTests.scala
+++ /dev/null
@@ -1,228 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License
- */
-
-package system
-
-import akka.testkit.TestProbe
-import org.apache.toree.communication.ZMQMessage
-import org.apache.toree.kernel.protocol.v5.kernel.{ActorLoader, Utilities}
-import Utilities._
-import org.apache.toree.kernel.protocol.v5
-import org.apache.toree.kernel.protocol.v5.{KernelMessage, SocketType, KMBuilder}
-import org.apache.toree.kernel.protocol.v5.content._
-import org.scalatest._
-import play.api.libs.json.Json
-import scala.concurrent.duration._
-import test.utils.NoArgSparkKernelTestKit
-
-
-/**
- * Used for system-wide (entire Spark Kernel) tests relating to Comm usage.
- *
- * @note Do not use non-generated target names or Comm ids! This is a shared
- * kernel instance (to avoid death by slowness) and it is not guaranteed
- * that previous tests will do proper cleanup!
- */
-@DoNotDiscover
-class TruncationTests
- extends NoArgSparkKernelTestKit
- with FunSpecLike with Matchers
-{
- private val MaxFishTime = 30.seconds
- import test.utils.SparkKernelDeployer._
-
- private def waitForExecuteReply(
- shell: TestProbe, headerId: v5.UUID, maxTime: Duration = MaxFishTime
- ): Unit =
- shell.fishForMessage(maxTime) {
- case KernelMessage(_, _, header, parentHeader, _, _)
- if header.msg_type == ExecuteReply.toTypeString &&
- parentHeader.msg_id == headerId=> true
- case _ => false
- }
-
- private def waitForBusyStatus(
- ioPub: TestProbe, headerId: v5.UUID, maxTime: Duration = MaxFishTime
- ): Unit =
- ioPub.fishForMessage(maxTime) {
- case KernelMessage(_, _, header, parentHeader, _, contentString)
- if header.msg_type == KernelStatus.toTypeString &&
- parentHeader.msg_id == headerId =>
- Json.parse(contentString).as[KernelStatus].execution_state == "busy"
- case _ => false
- }
-
- private def waitForIdleStatus(
- ioPub: TestProbe, headerId: v5.UUID, maxTime: Duration = MaxFishTime
- ): Unit =
- ioPub.fishForMessage(maxTime) {
- case KernelMessage(_, _, header, parentHeader, _, contentString)
- if header.msg_type == KernelStatus.toTypeString &&
- parentHeader.msg_id == headerId =>
- Json.parse(contentString).as[KernelStatus].execution_state == "idle"
- case _ => false
- }
-
- private def setupCallback(
- actorLoader: ActorLoader, shell: TestProbe, callbackFunc: String,
- targetName: String, expectedMessage: String
- ) = {
- // If open callback, there are four arguments
- // If msg/close callback, there are three arguments
- val callbackArgs =
- if (callbackFunc.toLowerCase.trim == "addopenhandler") """(_,_,_,_)"""
- else """(_,_,_)"""
-
- // Set up a callback (using stream messages as indicator)
- val codeRequest = s"""
- |kernel.comm.register("$targetName").$callbackFunc {
- | $callbackArgs => kernel.out.println("$expectedMessage")
- |}
- """.stripMargin.trim
- val (messageId, message) = buildZMQExecuteRequest(codeRequest)
- actorLoader.load(SocketType.Shell) ! message
-
- waitForExecuteReply(shell, messageId)
- }
-
- private def testForCallback(
- ioPub: TestProbe, expectedMessage: String, maxTime: Duration = MaxFishTime
- ) = {
- // Should discover an outgoing kernel message for stream
- ioPub.fishForMessage(maxTime) {
- case KernelMessage(_, _, header, _, _, contentString)
- if header.msg_type == StreamContent.toTypeString =>
- val callbackMessage =
- Json.parse(contentString).as[StreamContent].text.trim
- callbackMessage == expectedMessage
- case _ => false
- }
- }
-
- private def buildZMQExecuteRequest(codeRequest: String): (v5.UUID, ZMQMessage) = {
- val kernelMessage = KMBuilder()
- .withHeader(ExecuteRequest.toTypeString)
- .withContentString(ExecuteRequest(
- code = codeRequest,
- silent = false,
- store_history = false,
- user_expressions = v5.UserExpressions(),
- allow_stdin = true
- )).build
-
- (kernelMessage.header.msg_id, kernelMessage)
- }
-
- // Not using TestCommId to avoid problems if previous comm using that
- // id in tests was not properly closed
- private def buildZMQCommOpen(
- targetName: String, data: v5.MsgData,
- commId: v5.UUID = java.util.UUID.randomUUID().toString
- ): (v5.UUID, ZMQMessage) = {
- val kernelMessage = KMBuilder()
- .withHeader(CommOpen.toTypeString)
- .withContentString(CommOpen(
- comm_id = commId,
- target_name = targetName,
- data = data
- )).build
-
- (kernelMessage.header.msg_id, kernelMessage)
- }
-
- // Not using TestCommId to avoid problems if previous comm using that
- // id in tests was not properly closed
- private def buildZMQCommMsg(commId: v5.UUID, data: v5.MsgData): (v5.UUID, ZMQMessage) = {
- val kernelMessage = KMBuilder()
- .withHeader(CommMsg.toTypeString)
- .withContentString(CommMsg(
- comm_id = commId,
- data = data
- )).build
-
- (kernelMessage.header.msg_id, kernelMessage)
- }
-
- // Not using TestCommId to avoid problems if previous comm using that
- // id in tests was not properly closed
- private def buildZMQCommClose(commId: v5.UUID, data: v5.MsgData): (v5.UUID, ZMQMessage) = {
- val kernelMessage = KMBuilder()
- .withHeader(CommClose.toTypeString)
- .withContentString(CommClose(
- comm_id = commId,
- data = data
- )).build
-
- (kernelMessage.header.msg_id, kernelMessage)
- }
-
-
- private def executeCode(codeRequest : String, actorLoader : ActorLoader, ioPub: TestProbe ) : String = {
- val testTargetName = java.util.UUID.randomUUID().toString
-
-
- // Send our code to trigger the Comm API
- val (messageid, message) = buildZMQExecuteRequest(codeRequest)
- actorLoader.load(SocketType.Shell) ! message
-
- // Should discover an outgoing kernel message for comm_open
- val res= ioPub.fishForMessage(MaxFishTime) {
- case KernelMessage(_, _, header, parentHeader, _, _)
- if header.msg_type == ExecuteResult.toTypeString => true
- case _ => false
- }
- val kernelMessage=res.asInstanceOf[KernelMessage]
- val map=Json.parse(kernelMessage.contentString).as[ExecuteResult].data.asInstanceOf[Map[String,String]]
- val str= map.getOrElse("text/plain","NOT THERE")
-
- str
- }
-
- describe( "Test Truncation") {
- it("should show or not show types based on %showtypes") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
-
- executeCode("1",actorLoader,ioPub) should be ("1")
-
- executeCode(
- """%showtypes on
- |1
- """.stripMargin,actorLoader,ioPub) should be ("Int = 1")
-
- executeCode(
- """%showtypes off
- |1
- """.stripMargin,actorLoader,ioPub) should be ("1")
- }
- }
- it("should truncate or not truncate based on %truncate") {
- withNoArgSparkKernel { (actorLoader, heartbeat, shell, ioPub) =>
- executeCode("for ( a <- 1 to 300 ) yield a",actorLoader,ioPub) should endWith("...")
-
- executeCode(
- """%Truncation off
- |for ( a <- 1 to 300 ) yield a
- """.stripMargin,actorLoader,ioPub) should endWith("300)")
-
- executeCode(
- """%Truncation on
- |for ( a <- 1 to 300 ) yield a
- """.stripMargin,actorLoader,ioPub) should endWith("...")
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/a7f50032/kernel/src/test/scala/test/utils/NoArgSparkKernelTestKit.scala
----------------------------------------------------------------------
diff --git a/kernel/src/test/scala/test/utils/NoArgSparkKernelTestKit.scala b/kernel/src/test/scala/test/utils/NoArgSparkKernelTestKit.scala
deleted file mode 100644
index 3a06a21..0000000
--- a/kernel/src/test/scala/test/utils/NoArgSparkKernelTestKit.scala
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License
- */
-
-package test.utils
-
-import akka.testkit.TestKit
-
-class NoArgSparkKernelTestKit
- extends TestKit(SparkKernelDeployer.getNoArgSparkKernelActorSystem)
-{
- // Force initialization of no-arg Spark Kernel
- SparkKernelDeployer.noArgKernelBootstrap
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/a7f50032/kernel/src/test/scala/test/utils/SparkKernelDeployer.scala
----------------------------------------------------------------------
diff --git a/kernel/src/test/scala/test/utils/SparkKernelDeployer.scala b/kernel/src/test/scala/test/utils/SparkKernelDeployer.scala
deleted file mode 100644
index 62548c7..0000000
--- a/kernel/src/test/scala/test/utils/SparkKernelDeployer.scala
+++ /dev/null
@@ -1,186 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License
- */
-
-package test.utils
-
-import akka.actor.{Actor, ActorRef, ActorSystem, Props}
-import akka.testkit.TestProbe
-import com.typesafe.config.Config
-import org.apache.toree.boot.layer.{StandardBareInitialization, StandardComponentInitialization, StandardHandlerInitialization, StandardHookInitialization}
-import org.apache.toree.boot.{CommandLineOptions, KernelBootstrap}
-import org.apache.toree.kernel.protocol.v5.SocketType
-import org.apache.toree.kernel.protocol.v5.kernel.ActorLoader
-import org.apache.toree.kernel.protocol.v5.kernel.socket._
-import org.apache.toree.utils.LogLike
-import org.scalatest.mock.MockitoSugar
-import play.api.libs.json.Json
-
-/**
- * Represents an object that can deploy a singleton Spark Kernel for tests,
- * providing access to the actors used for socket communication.
- */
-object SparkKernelDeployer extends LogLike with MockitoSugar {
- private var actorSystem: ActorSystem = _
- private var actorLoader: ActorLoader = _
- private var heartbeatProbe: TestProbe = _
- private var heartbeatActor: ActorRef = _
- private var stdinProbe: TestProbe = _
- private var stdinActor: ActorRef = _
- private var shellProbe: TestProbe = _
- private var shellActor: ActorRef = _
- private var ioPubProbe: TestProbe = _
- private var ioPubActor: ActorRef = _
-
- private class ActorInterceptor(testProbe: TestProbe, actor: ActorRef)
- extends Actor
- {
- override def receive: Receive = {
- case m =>
- testProbe.ref.forward(m)
- actor ! m
- }
- }
-
-
-
- /**
- * Runs bare initialization, wrapping socket actors with test logic to
- * intercept messages.
- */
- private trait ExposedBareInitialization
- extends StandardBareInitialization with LogLike
- {
- override protected def createSockets(
- config: Config, actorSystem: ActorSystem, actorLoader: ActorLoader
- ): Unit =
- {
- logger.debug("Creating sockets")
-
- val socketConfig: SocketConfig = SocketConfig.fromConfig(config)
- logger.info("Connection Profile: "
- + Json.prettyPrint(Json.toJson(socketConfig)))
-
- logger.debug("Constructing ServerSocketFactory")
- val socketFactory = new SocketFactory(socketConfig)
-
- logger.debug("Initializing Heartbeat on port " +
- socketConfig.hb_port)
- val testHeartbeatActor = actorSystem.actorOf(
- Props(classOf[Heartbeat], socketFactory)
- )
-
- logger.debug("Initializing Stdin on port " +
- socketConfig.stdin_port)
- val testStdinActor = actorSystem.actorOf(
- Props(classOf[Stdin], socketFactory, actorLoader)
- )
-
- logger.debug("Initializing Shell on port " +
- socketConfig.shell_port)
- val testShellActor = actorSystem.actorOf(
- Props(classOf[Shell], socketFactory, actorLoader)
- )
-
- logger.debug("Initializing IOPub on port " +
- socketConfig.iopub_port)
- val testIOPubActor = actorSystem.actorOf(
- Props(classOf[IOPub], socketFactory)
- )
-
- // Update our publicly-available probes and create actors to wrap them
- heartbeatProbe = new TestProbe(actorSystem)
- val heartbeatInterceptor = actorSystem.actorOf(
- Props(new ActorInterceptor(heartbeatProbe, testHeartbeatActor)),
- name = SocketType.Heartbeat.toString
- )
- stdinProbe = new TestProbe(actorSystem)
- val stdinInterceptor = actorSystem.actorOf(
- Props(new ActorInterceptor(stdinProbe, testStdinActor)),
- name = SocketType.StdIn.toString
- )
- shellProbe = new TestProbe(actorSystem)
- val shellInterceptor = actorSystem.actorOf(
- Props(new ActorInterceptor(shellProbe, testShellActor)),
- name = SocketType.Shell.toString
- )
- ioPubProbe = new TestProbe(actorSystem)
- val ioPubInterceptor = actorSystem.actorOf(
- Props(new ActorInterceptor(ioPubProbe, testIOPubActor)),
- name = SocketType.IOPub.toString
- )
- }
-
- override protected def createActorLoader(actorSystem: ActorSystem) = {
- // Grab the created actor loader so we can use it with TestKit
- actorLoader = super.createActorLoader(actorSystem)
- actorLoader
- }
-
- override protected def createActorSystem(actorSystemName: String) = {
- // Grab the created actor system so we can use it with TestKit
- require(actorSystem != null,
- "Actor System not initialized before bootstrap!")
- //actorSystem = super.createActorSystem(actorSystemName)
- actorSystem
- }
- }
-
- /**
- * Represents the singleton KernelBootstrap instance that does not
- * receive any external commandline arguments.
- */
- lazy val noArgKernelBootstrap = {
- // Print out a message to indicate this fixture is being created
- logger.debug("Creating 'no external args' Spark Kernel through Kernel Bootstrap")
-
- val kernelBootstrap =
- (new KernelBootstrap(new CommandLineOptions(Nil).toConfig)
- with ExposedBareInitialization
- with StandardComponentInitialization
- with StandardHandlerInitialization
- with StandardHookInitialization).initialize()
-
- logger.debug("Finished initializing Kernel Bootstrap! Testing can now start!")
-
- kernelBootstrap
- }
-
- /**
- * Provides a gateway for tests to receive the "no external arguments"
- * KernelBootstrap through an actor loader and socket test probes.
- *
- * @param testCode The test code to execute
- * @return The results from the test code
- */
- def withNoArgSparkKernel(
- testCode: (ActorLoader, TestProbe, TestProbe, TestProbe) => Any
- ) = testCode(actorLoader, heartbeatProbe, shellProbe, ioPubProbe)
-
-
- /**
- * Retrieves the actor system for the "no external arguments" KernelBootstrap
- * instance. Will initialize the KernelBootstrap if the actor system is not
- * ready.
- *
- * @return The actor system of KernelBootstrap
- */
- def getNoArgSparkKernelActorSystem: ActorSystem =
- //if (actorSystem == null) { noArgKernelBootstrap; actorSystem }
- //else actorSystem
- if (actorSystem == null) { actorSystem = ActorSystem(); actorSystem }
- else actorSystem
-}