//设置package包名称以及导入依赖的类
package com.sky.kafka.configurator

import common.KafkaIntSpec
import kafka.admin.AdminUtils
import org.scalatest.concurrent.Eventually

import scala.util.Success

class KafkaConfiguratorFeatureSpec extends KafkaIntSpec with Eventually {

  "KafkaConfigurator" should "create new topics in Kafka from a file" in {
    val args = Array(
      "-f", getClass.getResource("/topic-configuration.yml").getPath,
      "--zookeeper", s"localhost:${kafkaServer.zookeeperPort.toString}"
    )
    val topics = List("topic1", "topic2")

    topics.map(AdminUtils.topicExists(zkUtils, _) shouldBe false)

    Main.run(args) shouldBe Success(())

    eventually {
      withClue("Topic exists: ") {
        topics.map(AdminUtils.topicExists(zkUtils, _) shouldBe true)
      }
    }
  }

} 

//设置package包名称以及导入依赖的类
package mesosphere.marathon
package integration.setup

import org.scalatest.concurrent.Eventually
import org.scalatest.time.{ Milliseconds, Span }

import scala.concurrent.duration._


object WaitTestSupport extends Eventually {
  def validFor(description: String, until: FiniteDuration)(valid: => Boolean): Boolean = {
    val deadLine = until.fromNow
    def checkValid(): Boolean = {
      if (!valid) throw new IllegalStateException(s"$description not valid for $until. Give up.")
      if (deadLine.isOverdue()) true else {
        Thread.sleep(100)
        checkValid()
      }
    }
    checkValid()
  }

  def waitUntil(description: String, maxWait: FiniteDuration)(fn: => Boolean): Unit = {
    eventually(timeout(Span(maxWait.toMillis, Milliseconds))) {
      if (!fn) throw new RuntimeException(s"$description not satisfied")
    }
  }
} 

//设置package包名称以及导入依赖的类
package dummy_authenticator.rest

import akka.http.scaladsl.model.StatusCodes.{CustomStatusCode, OK, Redirection, ServerError}
import akka.http.scaladsl.model.{ContentType, HttpEntity, Multipart, StatusCodes}
import akka.http.scaladsl.server.Route
import akka.http.scaladsl.testkit.ScalatestRouteTest
import dummy_authenticator.config.Configuration
import org.scalatest.concurrent.Eventually
import org.scalatest.{FunSpec, Matchers}

import scala.concurrent.duration._
import scala.language.postfixOps

trait RestTest extends FunSpec with Matchers with Eventually with ScalatestRouteTest with Routes {
  val config: Configuration = Configuration.get

  val BASE_PATH: String        = "/" + config.app.name
  val FILE_PATH: String        = BASE_PATH + "/share/file"
  val LIST_PATH: String        = BASE_PATH + "/share/filelist"
  val PATH_NONEXISTENT: String = "ShouldNotExist"
  val FILE_NONEXISTENT: String = "ShouldNotExist.file"
  val EMPTY_DIR: String        = "{\"keys\":[]}"

  
  def testPathStatus(path: String): Unit = {
    Get(path) ~> Route.seal(routes) ~> check {
      status should not be a[ServerError]
      status should not be a[Redirection]
      status should not be a[CustomStatusCode]
    }

    Delete(path) ~> Route.seal(routes) ~> check {
      status should not be a[ServerError]
      status should not be a[Redirection]
      status should not be a[CustomStatusCode]
    }

    Put(path) ~> Route.seal(routes) ~> check {
      status should not be a[ServerError]
      status should not be a[Redirection]
      status should not be a[CustomStatusCode]
    }

    Post(path) ~> Route.seal(routes) ~> check {
      status should not be a[ServerError]
      status should not be a[Redirection]
      status should not be a[CustomStatusCode]
    }
  }

  def setupFileForUpload(filename: String,
                         fileContent: String,
                         contentType: ContentType.NonBinary,
                         fieldName: String): Multipart.FormData.Strict = {
    Multipart.FormData(
      Multipart.FormData.BodyPart.Strict(fieldName, HttpEntity(contentType, fileContent), Map("filename" ? filename)))
  }
} 

//设置package包名称以及导入依赖的类
package co.horn.alkes.auth

import akka.http.scaladsl.model.HttpMethod
import akka.http.scaladsl.model.HttpMethods.{DELETE, GET, POST, PUT}
import akka.http.scaladsl.model.StatusCodes.{Forbidden, ServerError}
import akka.http.scaladsl.model.headers.{Authorization, OAuth2BearerToken}
import akka.http.scaladsl.server.Route
import akka.http.scaladsl.testkit.ScalatestRouteTest
import co.horn.alkes.config.Configuration
import co.horn.alkes.dao.DataHandler
import co.horn.alkes.dao.implementations.riak.RiakDataHandler
import co.horn.alkes.log.Logger
import co.horn.alkes.rest.Routes
import org.scalatest.{FunSpec, Matchers}
import org.scalatest.concurrent.Eventually

trait AuthTest extends FunSpec with Matchers with Eventually with ScalatestRouteTest with Routes {
  val config: Configuration = Configuration.get
  val dao: DataHandler      = new RiakDataHandler(config)
  val log: Logger           = config.log.tests

  // TODO: Define these all in just ONE spot. Need to keep DRY!
  val BASE_PATH: String  = "/" + config.app.name
  val FILE_PATH: String  = BASE_PATH + "/share/file"
  val LIST_PATH: String  = BASE_PATH + "/share/filelist"
  val META_PATH: String  = BASE_PATH + "/share/metadata"
  val THUMB_PATH: String = BASE_PATH + "/share/thumbnail"

  
  def canAccessRoute(token: OAuth2BearerToken, route: String, method: HttpMethod): Boolean = {
    method match {
      case GET =>
        Get(route).withHeaders(Authorization(token)) ~> Route.seal(routes) ~> check {
          status should not be a[ServerError]
          status != Forbidden
        }
      case PUT =>
        Put(route).withHeaders(Authorization(token)) ~> Route.seal(routes) ~> check {
          status should not be a[ServerError]
          status != Forbidden
        }
      case DELETE =>
        Get(route).withHeaders(Authorization(token)) ~> Route.seal(routes) ~> check {
          status should not be a[ServerError]
          status != Forbidden
        }
      case POST =>
        Post(route).withHeaders(Authorization(token)) ~> Route.seal(routes) ~> check {
          status should not be a[ServerError]
          status != Forbidden
        }
      case m => throw new IllegalArgumentException(s"$m is not an HttpMethod accepted by Alkes.")
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.tpalanga.account.route

import akka.http.scaladsl.model.StatusCodes
import akka.http.scaladsl.testkit.ScalatestRouteTest
import org.scalatest.concurrent.Eventually
import org.scalatest.{FlatSpec, Matchers}

object PingRouteSpec {
  trait Test {
    val pingRoute = new PingRoute
  }
}

class PingRouteSpec extends FlatSpec with ScalatestRouteTest with Matchers with Eventually {
  import PingRouteSpec._

  "PingRoute" should "respond to GET" in new Test {
    Get("/ping") ~> pingRoute.route ~> check {
      eventually {
        status shouldBe StatusCodes.OK
      }
      responseAs[String] shouldBe "<h1>pong</h1>"
    }
  }

} 

//设置package包名称以及导入依赖的类
package org.cristal.repository

import akka.actor.ActorSystem
import akka.testkit.{ImplicitSender, TestKit}
import akka.util.Timeout
import com.github.t3hnar.bcrypt._
import org.cristal.model.{NewUser, User}
import org.cristal.repository.UserRepository.UserCreated
import org.cristal.repository.dao.UserDAO
import org.mockito.ArgumentMatchers._
import org.mockito.Mockito.when
import org.scalatest.concurrent.Eventually
import org.scalatest.mockito.MockitoSugar
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
import scala.concurrent.Future
import scala.concurrent.duration._

class UserRepositorySpec(_system: ActorSystem) extends TestKit(_system) with ImplicitSender
  with WordSpecLike with Matchers with BeforeAndAfterAll with Eventually with MockitoSugar {
  implicit val duration: Timeout = 10 seconds
  def this() = this(ActorSystem("UserRepositorySpecSystem"))

  override def afterAll {
    TestKit.shutdownActorSystem(system)
  }

  "An UserRepository Actor" should {
    "Create a new user" in {
      val userDAO = mock[UserDAO]
      val passowod = "my_password"
      val encryptedPassword = passowod.bcrypt
      when(userDAO.insert(any())).thenReturn(Future.successful(()))
      val userRepository = system.actorOf(UserRepository.props(userDAO))
      val newUser = NewUser("name", passowod, "[email protected]", "John", "Doe")
      userRepository ! UserRepository.CreateUser(newUser, self)
      expectMsgPF() {
        case UserCreated(User("name", encryptedPassword, "[email protected]", "John", "Doe")) => ()
      }
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.coding42.gol

import org.scalatest.{FlatSpec, Matchers}
import org.scalatest.concurrent.{Eventually, IntegrationPatience}

import scalafx.application.Platform

class LifeAppSpec extends FlatSpec with Eventually with IntegrationPatience with Matchers {

  it should "open the app and create a default canvas of 300x200" in {
    runApp()
    eventually {
      LifeApp.canvas.width.value shouldBe 300
      LifeApp.canvas.height.value shouldBe 200
    }
  }

  it should "check app updates steps" in {
    eventually {
      step shouldBe "Step: 1"
    }

    eventually {
      step shouldBe "Step: 2"
    }
  }

  it should "restart button does start steps again an can change size" in {
    eventually {
      step shouldBe "Step: 3"
    }

    Platform.runLater(LifeApp.restartBtn.fire())
    Platform.runLater(LifeApp.widthField.text = "100")
    Platform.runLater(LifeApp.heightField.text = "50")
    Platform.runLater(LifeApp.okBtn.fire())

    eventually {
      step shouldBe "Step: 1"
      LifeApp.canvas.width.value shouldBe 100
      LifeApp.canvas.height.value shouldBe 50
    }
  }

  private def step = LifeApp.stepText.text.value

  private def runApp(): Unit = {
    val thread = new Thread {
      override def run(): Unit = {
        LifeApp.main(Array.empty)
      }
    }
    thread.start()
  }

} 

//设置package包名称以及导入依赖的类
package de.choffmeister.microserviceutils

import akka.Done
import akka.actor.{ActorSystem, Props}
import akka.stream.scaladsl.{Concat, Keep, Sink, Source}
import akka.stream.{KillSwitches, ThrottleMode}
import akka.testkit.{TestKit, TestProbe}
import org.scalatest._
import org.scalatest.concurrent.Eventually
import org.scalatest.time.{Millis, Seconds, Span}

import scala.collection.mutable.ListBuffer
import scala.concurrent.Future
import scala.concurrent.duration._

class PerpetualStreamTest extends TestKit(ActorSystem()) with FlatSpecLike with Matchers with Eventually with BeforeAndAfterAll {
  implicit val defaultPatience = PatienceConfig(timeout = Span(30, Seconds), interval = Span(100, Millis))

  "PerpetualStream" should "work" in {
    val list = ListBuffer.empty[Int]
    val stream = system.actorOf(Props(new NumberGeneratingPerpetualStream(list)))
    val probe = new TestProbe(system)
    probe.watch(stream)

    eventually(require(list.length >= 20, "Must have emitted at least 20 numbers"))
    list.take(20) should be((-10 until 10).toList)

    GracefulShutdownExtension(system).triggerGracefulShutdown()
    probe.expectTerminated(stream)
  }

  override def afterAll(): Unit = TestKit.shutdownActorSystem(system)

  class NumberGeneratingPerpetualStream(list: ListBuffer[Int]) extends PerpetualStream {
    var count = 0

    override def stream = {
      val currentCount = count
      count = count + 1

      val source = currentCount match {
        case 0 => Source.combine(Source(-10 until 0), Source.failed[Int](new RuntimeException("ERROR")))(Concat(_))
        case 1 => Source(0 until 10)
        case _ => Source(10 to 100000).throttle(1000, 100.millis, 100, ThrottleMode.shaping)
      }

      source
        .via(watcher)
        .viaMat(KillSwitches.single)(Keep.right)
        .map { i => list.append(i); i }
        .to(Sink.ignore)
        .mapMaterializedValue(ks => () => {
          ks.shutdown()
          Future.successful(Done)
        })
    }
  }
} 

//设置package包名称以及导入依赖的类
package uk.gov.hmrc.agentinvitations.scenarios

import org.scalatest._
import org.scalatest.concurrent.Eventually
import uk.gov.hmrc.agentinvitations.support._
import uk.gov.hmrc.domain.{AgentCode, Nino}


class AgencyFiltersByClientIdAndStatusPlatformISpec extends FeatureSpec with ScenarioHelpers with GivenWhenThen with Matchers with MongoAppAndStubs with Inspectors with Inside with Eventually {

  implicit val arn = RandomArn()
  private implicit val agentCode = AgentCode("LMNOP123456")
  val nino: Nino = nextNino
  val nino2: Nino = nextNino

  feature("Agencies can filter")  {

    scenario("on the client id and status of invitations") {
      val agency = new AgencyRequests(this, arn, port)
      val client = new ClientRequests(this, nino, port)
      Given("An agent is logged in")
      given().agentAdmin(arn, agentCode).isLoggedInWithSessionId().andIsSubscribedToAgentServices()
      given().client(clientId = nino).isLoggedInWithSessionId().hasABusinessPartnerRecord().aRelationshipIsCreatedWith(arn)
      given().client(clientId = nino2).hasABusinessPartnerRecord()

      When("An agent sends invitations to Client 1")
      agencySendsSeveralInvitations(agency)(
        (nino, MtdItService),
        (nino, MtdItService)
      )

      And("Sends an invitations to Client 2")
      agency sendInvitation(nino2, MtdItService)

      And("Client 1 accepts the first invitation")
      clientAcceptsFirstInvitation(client)

      Then("The agent filters by Client 1 and Pending")
      agencyFiltersByClient1Pending(agency)

      Then("The agent filters by Client 2 and Accepted")
      agencyFiltersByClient2Accepted(agency)
    }
  }

  def agencyFiltersByClient1Pending(agency: AgencyRequests) = {
    val invitations = agency.sentInvitations(filteredBy = Seq("clientId" -> nino.value, "status" -> "Pending"))

    invitations.numberOfInvitations shouldBe 1
  }

  def agencyFiltersByClient2Accepted(agency: AgencyRequests) = {
    val invitations = agency.sentInvitations(filteredBy = Seq("clientId" -> nino2.value, "status" -> "Accepted"))

    invitations.numberOfInvitations shouldBe 0
  }
} 

//设置package包名称以及导入依赖的类
package uk.gov.hmrc.agentinvitations.scenarios

import org.scalatest._
import org.scalatest.concurrent.Eventually
import uk.gov.hmrc.agentinvitations.support._
import uk.gov.hmrc.domain.{AgentCode, Nino}


class AgencyFilteringByClientIdIPlatformISpec extends FeatureSpec with ScenarioHelpers with GivenWhenThen with Matchers with MongoAppAndStubs with Inspectors with Inside with Eventually {

  override val arn = RandomArn()
  override val nino: Nino = nextNino

  private implicit val agentCode = AgentCode("LMNOP123456")
  private val nino2: Nino = nextNino

  feature("Agencies can filter")  {

    scenario("on the status of clients invitations") {
      val agency = new AgencyRequests(this, arn, port)
      val client1 = new ClientRequests(this, nino, port)
      val client2 = new ClientRequests(this, nino2, port)

      Given("An agent is logged in")
      given().agentAdmin(arn, agentCode).isLoggedInWithSessionId().andIsSubscribedToAgentServices()
      given().client(clientId = nino).hasABusinessPartnerRecord()
      given().client(clientId = nino2).hasABusinessPartnerRecord()

      And("the Agency has sent 1 invitation to 2 different clients")
      agencySendsSeveralInvitations(agency)(
        (nino, MtdItService),
        (nino2, MtdItService)
      )

      When(s"the Agency filters by client ID")
      Then(s"only the client matching that id is returned")
      agencyFiltersById(agency, client1.clientId)
      agencyFiltersById(agency, client2.clientId)
    }
  }

  private def agencyFiltersById(agency: AgencyRequests, clientId: Nino): Unit = {
    val invitation = agency.sentInvitations(filteredBy = Seq("clientId" -> clientId.value))
    invitation.numberOfInvitations shouldBe 1
    invitation.firstInvitation.status shouldBe "Pending"
    invitation.firstInvitation.arn shouldBe agency.arn
    invitation.firstInvitation.clientId shouldBe clientId
  }
} 

//设置package包名称以及导入依赖的类
package uk.gov.hmrc.agentinvitations.scenarios

import org.scalatest._
import org.scalatest.concurrent.Eventually
import uk.gov.hmrc.agentinvitations.support.{AgencyRequests, MongoAppAndStubs, RandomArn, ScenarioHelpers}
import uk.gov.hmrc.domain.{AgentCode, Nino}


class AgencyFiltersByClientIdTypePlatformISpec extends FeatureSpec with ScenarioHelpers with GivenWhenThen with Matchers with MongoAppAndStubs with Inspectors with Inside with Eventually {

  implicit val arn = RandomArn()
  private implicit val agentCode = AgentCode("LMNOP123456")
  val nino: Nino = nextNino

  feature("Agencies can filter")  {

    scenario("on the clientIdType of invitations") {
      val agency = new AgencyRequests(this, arn, port)
      Given("An agent is logged in")
      given().agentAdmin(arn, agentCode).isLoggedInWithSessionId().andIsSubscribedToAgentServices()
      given().client(clientId = nino).hasABusinessPartnerRecord()

      When("An agent sends several invitations")
      agencySendsSeveralInvitations(agency)(
        (nino, MtdItService),
        (nino, MtdItService)
      )

      Then("The agent filters by clientIdType=ni")
      agencyFiltersByNi(agency)

      Then("The agent filters by clientIdType=mtd-it-id")
      agencyFiltersByMtdItId(agency)
    }
  }

  def agencyFiltersByNi(agency: AgencyRequests) = {
    val invitations = agency.sentInvitations(filteredBy = Seq("clientIdType" -> "ni"))

    invitations.numberOfInvitations shouldBe 2
  }

  def agencyFiltersByMtdItId(agency: AgencyRequests) = {
    val invitations = agency.sentInvitations(filteredBy = Seq("clientIdType" -> "mtd-it-id"))

    invitations.numberOfInvitations shouldBe 0
  }
} 

//设置package包名称以及导入依赖的类
package uk.gov.hmrc.agentinvitations.scenarios

import org.scalatest.{Inside, Inspectors}
import org.scalatest.concurrent.Eventually
import uk.gov.hmrc.agentinvitations.support._
import uk.gov.hmrc.domain.{AgentCode, Nino}
import uk.gov.hmrc.play.test.UnitSpec

class NoInvitationsPlatformISpec extends UnitSpec with MongoAppAndStubs with Inspectors with Inside with Eventually with Requests {

  private implicit val arn = RandomArn()
  private implicit val agentCode = AgentCode("LMNOP123456")
  private val nino: Nino = nextNino

  "Before the Agency has sent any invitations" in {
    val agency = new AgencyRequests(this, arn, port)
    val client = new ClientRequests(this, nino, port)

    given().agentAdmin(arn, agentCode).isLoggedInWithSessionId().andIsSubscribedToAgentServices()
    given().client(clientId = nino).isLoggedInWithSessionId()

    info("the Agency sent invitations should be empty")
    val agencyResponse = agency.sentInvitations()
    agencyResponse.numberOfInvitations shouldBe 0
    agencyResponse.links.invitations shouldBe 'empty
    agencyResponse.links.selfLink shouldBe s"/agent-invitations/agencies/${arn.value}/invitations/sent"
    agencyResponse.embedded.isEmpty shouldBe true

    info("the Clients received invitations should be empty")
    val clientResponse = client.getInvitations()
    clientResponse.numberOfInvitations shouldBe 0
    clientResponse.links.invitations shouldBe 'empty
    clientResponse.links.selfLink shouldBe s"/agent-invitations/clients/ni/${nino.value}/invitations/received"
    clientResponse.embedded.isEmpty shouldBe true
  }
} 

//设置package包名称以及导入依赖的类
package com.mnubo.flink.streaming.connectors
package elasticsearch

import org.scalatest.concurrent.Eventually
import org.scalatest.time.SpanSugar

import scala.sys.process._
import scala.util.Try

class ElasticsearchTestServer(version: String, isClusterGreen: (String, Int) => Boolean) extends AutoCloseable with Eventually with SpanSugar {
  private val hasRecoveredIndicesStateRegex = """recovered \[\d+\] indices into cluster_state""".r
  val host = {
    val hostVar = System.getenv("DOCKER_HOST")
    if (hostVar != null)
      """\d+\.[0-9\.]+""".r
        .findFirstIn(hostVar)
        .getOrElse("127.0.0.1")
    else
      "127.0.0.1"
  }
  val containerId = s"docker run -d -P elasticsearch:$version --network.publish_host $host".!!.trim
  val httpPort = esPort(9200)
  val esTransportPort = esPort(9300)

  eventually(timeout(20.seconds), interval(500.millis)) {
    require(hasRecoveredIndicesState && isClusterGreen(host, esTransportPort), "ES Still not started...")
  }

  override def close() = {
    Try(s"docker stop $containerId".!)
    Try(s"docker rm $containerId".!)
  }

  private def hasRecoveredIndicesState = {
    val logs = s"docker logs $containerId".!!
    hasRecoveredIndicesStateRegex.findFirstIn(logs).isDefined
  }

  private def esPort(exposedPort: Int) = Seq(
    "docker",
    "inspect",
    s"""--format='{{(index (index .NetworkSettings.Ports "$exposedPort/tcp") 0).HostPort}}'""",
    containerId
  ).!!.trim.toInt
} 

//设置package包名称以及导入依赖的类
package com.octo.nad.handson.spark.specs

import com.octo.nad.handson.spark.utils.AppConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import org.scalatest.concurrent.Eventually
import org.scalatest.{Matchers, BeforeAndAfter, FlatSpec}

trait SparkStreamingSpec extends FlatSpec with BeforeAndAfter with Eventually with Matchers with AppConf{
  private var _ssc: StreamingContext = _
  private var _sc: SparkContext = _
  private val appName = this.getClass.getSimpleName+System.currentTimeMillis()
  private val master = "local[*]"
  private val batchWindow = Seconds(1)
  def ssc = _ssc
  def sc = _sc

  def sparkConf(appName: String, master: String) : SparkConf = {
    new SparkConf()
      .setAppName(appName)
      .setMaster(master)
      .set("spark.cassandra.connection.host", CassandraHostName)
      .set("spark.driver.allowMultipleContexts", true.toString)

  }
  before{
    _sc = new SparkContext(sparkConf(appName,master))
    _sc.setLogLevel("WARN")
    _ssc = new StreamingContext(_sc,batchWindow)

  }

  after{
    if(_ssc != null) {
      _ssc.stop()
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.octo.nad.handson.spark.specs

import com.octo.nad.handson.spark.utils.AppConf
import org.apache.spark.{SparkConf, SparkContext}
import org.scalatest.concurrent.Eventually
import org.scalatest.{Matchers, BeforeAndAfter, FlatSpec}

trait SparkBatchSpec extends FlatSpec with BeforeAndAfter with Eventually with Matchers with AppConf{
  private var _sc: SparkContext = _
  private val appName = this.getClass.getSimpleName+System.currentTimeMillis()
  private val master = "local[*]"
  def sc = _sc

  def sparkConf(appName: String, master: String) : SparkConf = {
    new SparkConf()
      .setAppName(appName)
      .setMaster(master)
      .set("spark.cassandra.connection.host", CassandraHostName)
      .set("spark.driver.allowMultipleContexts", true.toString)

  }
  before{
    _sc = new SparkContext(sparkConf(appName,master))
    _sc.setLogLevel("WARN")
  }

  after{
    if (_sc != null) _sc.stop()
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.util

import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.concurrent.{Eventually, IntegrationPatience}
import org.scalatest.junit.{AssertionsForJUnit, JUnitRunner}
import org.scalatest.mock.MockitoSugar

@RunWith(classOf[JUnitRunner])
class ExitGuardTest
  extends FunSuite
  with MockitoSugar
  with AssertionsForJUnit
  with Eventually
  with IntegrationPatience {

  test("guard creates thread, unguard removes") {
    val name = s"ExitGuardTest-${System.nanoTime}"
    val guard = ExitGuard.guard(name)
    val (thread, guards) = ExitGuard.guards.get

    assert(!thread.isDaemon)
    assert(thread.isAlive)
    assert(guards.map(_.reason).contains(name))

    guard.unguard()

    // depending on what has been registered and unregistered,
    // either there should be no guards or our name should not be in the list.
    ExitGuard.guards match {
      case None =>
        eventually { assert(!thread.isAlive, ExitGuard.explainGuards()) }
      case Some((_, gs)) =>
        assert(!gs.map(_.reason).contains(name))
    }
  }

  test("explain shows reason") {
    val guard = ExitGuard.guard("<%= reason %>")
    assert(ExitGuard.explainGuards().contains("<%= reason %>"))
    guard.unguard()
    assert(!ExitGuard.explainGuards().contains("<%= reason %>"))
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.mux

import com.twitter.concurrent.AsyncQueue
import com.twitter.conversions.time._
import com.twitter.finagle.mux.transport.Message
import com.twitter.finagle.stats.NullStatsReceiver
import com.twitter.finagle.transport.QueueTransport
import com.twitter.finagle.{Path, Service}
import com.twitter.io.Buf
import com.twitter.util.{Await, Promise}
import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.concurrent.{Eventually, IntegrationPatience}
import org.scalatest.junit.JUnitRunner

@RunWith(classOf[JUnitRunner])
class DispatcherTest extends FunSuite
  with Eventually
  with IntegrationPatience {

  test("Discard request properly sent") {
    @volatile var handled = false
    val p = Promise[Response]()
    p.setInterruptHandler { case t: Throwable =>
      handled = true
    }

    val svc = Service.mk[Request, Response](_ => p)

    val q0, q1 = new AsyncQueue[Message]
    val clientTrans = new QueueTransport[Message, Message](q0, q1)
    val serverTrans = new QueueTransport[Message, Message](q1, q0)

    val server = ServerDispatcher.newRequestResponse(serverTrans, svc)
    val session = new ClientSession(
      clientTrans, FailureDetector.NullConfig, "test", NullStatsReceiver)
    val client = ClientDispatcher.newRequestResponse(session)

    val f = client(Request(Path.empty, Buf.Empty))
    assert(!f.isDefined)
    assert(!p.isDefined)
    f.raise(new Exception())
    eventually { assert(handled) }
    Await.ready(server.close().join(client.close()), 5.seconds)
  }
} 

//设置package包名称以及导入依赖的类
import org.scalatest.concurrent.{Eventually, IntegrationPatience}
import org.scalatest.{MustMatchers, OptionValues, fixture}
import org.scalatestplus.play.{PortNumber, WsScalaTestClient}
import play.api.libs.ws.{WSClient, WSRequest}
import play.api.mvc.Call

abstract class MixedPlaySpecWithNoDefaultApp extends fixture.WordSpec
  with MustMatchers
  with OptionValues
  with MixedFixturesWithNoDefaultApp
  with Eventually
  with IntegrationPatience
  with WsScalaTestClient
{

  //def wsCall(call: Call)(implicit portNumber: PortNumber, wsClient: WSClient): WSRequest = doCall(call.url, wsClient, portNumber)

  // def wsUrl(url: String)(implicit portNumber: PortNumber, wsClient: WSClient): WSRequest = doCall(url, wsClient, portNumber)

  //private def doCall(url: String, wsClient: WSClient, portNumber: PortNumber) = {
  //  wsClient.url("http://localhost:" + portNumber.value + url)
  //}
} 

//设置package包名称以及导入依赖的类
package com.github.dnvriend

import akka.actor.{ActorRef, ActorSystem, PoisonPill}
import akka.stream.Materializer
import akka.testkit.TestProbe
import akka.util.Timeout
import org.scalatest._
import org.scalatest.concurrent.{Eventually, ScalaFutures}
import org.scalatestplus.play.guice.GuiceOneServerPerSuite
import play.api.test.WsTestClient

import scala.concurrent.ExecutionContext
import scala.concurrent.duration._
import scala.reflect.ClassTag

class IntegrationSpec extends FlatSpec
  with Matchers
  with GivenWhenThen
  with OptionValues
  with TryValues
  with ScalaFutures
  with WsTestClient
  with BeforeAndAfterAll
  with BeforeAndAfterEach
  with Eventually
  with GuiceOneServerPerSuite {

  def getComponent[A: ClassTag]: A = app.injector.instanceOf[A]

  // set the port number of the HTTP server
  override lazy val port: Int = 9001
  implicit val pc: PatienceConfig = PatienceConfig(timeout = 30.seconds, interval = 300.millis)
  implicit val system: ActorSystem = getComponent[ActorSystem]
  implicit val ec: ExecutionContext = getComponent[ExecutionContext]
  implicit val mat: Materializer = getComponent[Materializer]
  implicit val timeout: Timeout = 10.seconds

  def killActors(actors: ActorRef*): Unit = {
    val tp = TestProbe()
    actors.foreach { (actor: ActorRef) =>
      tp watch actor
      actor ! PoisonPill
      tp.expectTerminated(actor)
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.beachape.sparkka

import akka.actor.ActorSystem
import akka.stream.{ SourceShape, ActorMaterializer }
import akka.stream.scaladsl._
import org.scalatest.time.{ Millis, Seconds, Span }
import org.scalatest.{ Matchers, FunSpec }
import org.scalatest.concurrent.{ Eventually, ScalaFutures }


class StreamingSpec extends FunSpec with ScalaFutures with Matchers with Eventually {

  override implicit val patienceConfig: PatienceConfig = PatienceConfig(Span(30, Seconds), Span(150, Millis))

  describe("streamConnection") {

    it("should allow both the original flow and the connected InputDStream to receive all expected values") {
      implicit val actorSystem = ActorSystem()
      implicit val materializer = ActorMaterializer()
      implicit val ssc = LocalContext.ssc

      // InputDStream can then be used to build elements of the graph that require integration with Spark
      val (inputDStream, feedDInput) = Streaming.connection[Int]()
      val source = Source.fromGraph(GraphDSL.create() { implicit builder =>

        import GraphDSL.Implicits._

        val source = Source(1 to 10)

        val bCast = builder.add(Broadcast[Int](2))
        val merge = builder.add(Merge[Int](2))

        val add1 = Flow[Int].map(_ + 1)
        val times3 = Flow[Int].map(_ * 3)
        source ~> bCast ~> add1 ~> merge
        bCast ~> times3 ~> feedDInput ~> merge

        SourceShape(merge.out)
      })

      val reducedFlow = source.runWith(Sink.fold(0)(_ + _))
      whenReady(reducedFlow)(_ shouldBe 230)

      val sharedVar = ssc.sparkContext.accumulator(0)
      inputDStream.foreachRDD { rdd =>
        rdd.foreach { i =>
          sharedVar += i
        }
      }
      ssc.start()
      eventually(sharedVar.value shouldBe 165)
    }

  }

}