//设置package包名称以及导入依赖的类
package org.zalando.hutmann.spec

import org.scalatest.concurrent.PatienceConfiguration.Interval
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.time.{ Seconds, Span }
import org.scalatestplus.play.{ PlaySpec, PortNumber, WsScalaTestClient }
import play.api.Application
import play.api.http.{ HeaderNames, MimeTypes }
import play.api.libs.ws.{ WSClient, WSResponse }

trait PlayUnitSpec extends PlaySpec with ScalaFutures with WsScalaTestClient {
  def myPublicAddress(): String
  implicit val portNumber: PortNumber

  def callWs(testPaymentGatewayURL: String)(implicit app: Application): WSResponse = {
    implicit val wSClient = app.injector.instanceOf[WSClient]
    val callbackURL = s"http://${myPublicAddress()}/callback"
    whenReady(
      wsUrl(testPaymentGatewayURL)
        .withQueryStringParameters("callbackURL" -> callbackURL)
        .withHttpHeaders(HeaderNames.ACCEPT -> MimeTypes.TEXT)
        .get(),
      Interval(Span(10, Seconds))
    ) { result =>
        result
      }
  }
} 

//设置package包名称以及导入依赖的类
import org.scalatest._
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.time.{Seconds, Span}
import slick.driver.H2Driver.api._
import slick.jdbc.meta._

class TablesSuite extends FunSuite with BeforeAndAfter with ScalaFutures {
  implicit override val patienceConfig = PatienceConfig(timeout = Span(5, Seconds))

  val suppliers = TableQuery[Suppliers]
  val coffees = TableQuery[Coffees]
  
  var db: Database = _

  def createSchema() =
    db.run((suppliers.schema ++ coffees.schema).create).futureValue
  
  def insertSupplier(): Int =
    db.run(suppliers += (101, "Acme, Inc.", "99 Market Street", "Groundsville", "CA", "95199")).futureValue
  
  before { db = Database.forConfig("h2mem1") }
  
  test("Creating the Schema works") {
    createSchema()
    
    val tables = db.run(MTable.getTables).futureValue

    assert(tables.size == 2)
    assert(tables.count(_.name.name.equalsIgnoreCase("suppliers")) == 1)
    assert(tables.count(_.name.name.equalsIgnoreCase("coffees")) == 1)
  }

  test("Inserting a Supplier works") {
    createSchema()
    
    val insertCount = insertSupplier()
    assert(insertCount == 1)
  }
  
  test("Query Suppliers works") {
    createSchema()
    insertSupplier()
    val results = db.run(suppliers.result).futureValue
    assert(results.size == 1)
    assert(results.head._1 == 101)
  }
  
  after { db.close }
} 

//设置package包名称以及导入依赖的类
package akka.stream.alpakka.s3.impl

import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, ActorMaterializerSettings}
import akka.stream.scaladsl.{Sink, Source}
import akka.testkit.TestKit
import akka.util.ByteString
import org.scalatest.time.{Millis, Seconds, Span}
import org.scalatest.{BeforeAndAfterAll, FlatSpecLike, Matchers}
import org.scalatest.concurrent.ScalaFutures

class MemoryBufferSpec(_system: ActorSystem)
    extends TestKit(_system)
    with FlatSpecLike
    with Matchers
    with BeforeAndAfterAll
    with ScalaFutures {

  def this() = this(ActorSystem("MemoryBufferSpec"))

  implicit val defaultPatience =
    PatienceConfig(timeout = Span(5, Seconds), interval = Span(30, Millis))

  implicit val materializer = ActorMaterializer(ActorMaterializerSettings(system).withDebugLogging(true))

  "MemoryBuffer" should "emit a chunk on its output containg the concatenation of all input values" in {
    val result = Source(Vector(ByteString(1, 2, 3, 4, 5), ByteString(6, 7, 8, 9, 10, 11, 12), ByteString(13, 14)))
      .via(new MemoryBuffer(200))
      .runWith(Sink.seq)
      .futureValue

    result should have size (1)
    val chunk = result.head
    chunk.size should be(14)
    chunk.data.runWith(Sink.seq).futureValue should be(Seq(ByteString(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14)))
  }

  it should "fail if more than maxSize bytes are fed into it" in {
    whenReady(
      Source(Vector(ByteString(1, 2, 3, 4, 5), ByteString(6, 7, 8, 9, 10, 11, 12), ByteString(13, 14)))
        .via(new MemoryBuffer(10))
        .runWith(Sink.seq)
        .failed
    ) { e =>
      e shouldBe a[IllegalStateException]
    }
  }
} 

//设置package包名称以及导入依赖的类
package akka.stream.alpakka.s3.impl

import akka.testkit.TestKit
import akka.stream.ActorMaterializerSettings
import org.scalatest.BeforeAndAfterAll
import org.scalatest.concurrent.ScalaFutures
import akka.stream.ActorMaterializer
import akka.actor.ActorSystem
import org.scalatest.Matchers
import org.scalatest.FlatSpecLike
import akka.stream.scaladsl.Source
import akka.stream.scaladsl.Flow
import akka.util.ByteString
import akka.stream.scaladsl.Sink
import org.scalatest.time.{Millis, Seconds, Span}
import scala.concurrent.duration._

class SplitAfterSizeSpec(_system: ActorSystem)
    extends TestKit(_system)
    with FlatSpecLike
    with Matchers
    with BeforeAndAfterAll
    with ScalaFutures {

  def this() = this(ActorSystem("SplitAfterSizeSpec"))
  implicit val defaultPatience =
    PatienceConfig(timeout = Span(5, Seconds), interval = Span(30, Millis))

  implicit val materializer = ActorMaterializer(ActorMaterializerSettings(system).withDebugLogging(true))

  "SplitAfterSize" should "yield a single empty substream on no input" in {
    Source
      .empty[ByteString]
      .via(
        SplitAfterSize(10)(Flow[ByteString]).concatSubstreams
      )
      .runWith(Sink.seq)
      .futureValue should be(Seq.empty)
  }

  it should "start a new stream after the element that makes it reach a maximum, but not split the element itself" in {
    Source(Vector(ByteString(1, 2, 3, 4, 5), ByteString(6, 7, 8, 9, 10, 11, 12), ByteString(13, 14)))
      .via(
        SplitAfterSize(10)(Flow[ByteString]).prefixAndTail(10).map { case (prefix, tail) => prefix }.concatSubstreams
      )
      .runWith(Sink.seq)
      .futureValue should be(
      Seq(
        Seq(ByteString(1, 2, 3, 4, 5), ByteString(6, 7, 8, 9, 10, 11, 12)),
        Seq(ByteString(13, 14))
      )
    )
  }

} 

//设置package包名称以及导入依赖的类
package uk.co.appministry.scathon.client

import org.scalatest.concurrent._
import org.scalatest.time.{Millis, Seconds, Span}
import org.scalatest.{BeforeAndAfterAll, Inside, Matchers, WordSpec}
import uk.co.appministry.scathon.testServer.TestMarathon

class TestBase extends WordSpec
  with Eventually
  with ScalaFutures
  with Inside
  with Matchers
  with BeforeAndAfterAll
  with AsyncAssertions {

  implicit override val patienceConfig = PatienceConfig(timeout = scaled(Span(10, Seconds)), interval = scaled(Span(100, Millis)))

  var client: Client = _
  var server: TestMarathon = _

  override def beforeAll: Unit = {
    server = new TestMarathon
    server.start()
    client = new Client(port = server.port.get)
  }

  override def afterAll: Unit = {
    server.stop()
  }

} 

//设置package包名称以及导入依赖的类
package practice

import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.exceptions.{TestCanceledException, TestFailedDueToTimeoutException}
import org.scalatest.junit.JUnitRunner
import org.scalatest.time.{Millis, Span, Seconds}
import org.scalatest.concurrent.Timeouts._




@RunWith(classOf[JUnitRunner])
class SimulationSuite extends FunSuite{
  object sim extends Circuits with Parameters
  import sim._

  test("a simple test"){
    val in1, in2, sum, carry = new Wire
    probe("sum", sum)
    probe("carry", carry)
    halfAdder(in1, in2, sum, carry)
    in1.setSignal(true)
    sim.run()
    in2.setSignal(true)
    sim.run()
    assert(carry.getSignal == true)
    assert(sum.getSignal == false)
  }

  test("will not terminate") {
    try {
      cancelAfter(Span(10, Seconds)) {
        val in3 = new Wire
        probe("in3", in3)
        inverter(in3, in3)
        sim.run()
      }
    }
    catch {
      case x:TestCanceledException =>
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.piotrglazar.receiptlottery.core

import com.piotrglazar.receiptlottery.Token
import com.piotrglazar.receiptlottery.utils.ScalajHttpAdapter
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.prop.TableDrivenPropertyChecks._
import org.scalatest.time.{Seconds, Span}
import org.scalatest.{FlatSpec, Matchers}

import scala.concurrent.ExecutionContext.Implicits.global

class ResultFetcherTest extends FlatSpec with Matchers with ScalaFutures {

  implicit val timeout = PatienceConfig(Span(1, Seconds))

  val resultFetcher = new ResultFetcher("https://loteriaparagonowa.gov.pl/wyniki", new ScalajHttpAdapter(2000), global)

  val tokens = Table(("token", "result"),
    (Token("D2T1UGL9M34"), true),
    (Token("C91B2MGBM5F"), false))

  "ResultFetcher" should "find result for token" in {
    forAll(tokens) { (token: Token, expectedResult: Boolean) =>
      // when
      val result = resultFetcher.hasResult(token)

      // then
      result.futureValue shouldBe expectedResult
    }
  }

} 

//设置package包名称以及导入依赖的类
package ru.tolsi.matcher.naive

import scala.concurrent.Future
import org.scalatest.concurrent.PatienceConfiguration.Timeout
import org.scalatest.time.{Seconds, Span}
import ru.tolsi.matcher.{ClientInfo, Order, OrderType, ReverseOrders, UnitSpec}

class SingleThreadOrderExecutorSpec extends UnitSpec {
  describe("execute method") {
    implicit val ec = scala.concurrent.ExecutionContext.global
    it("should execure reverse orders on users from repo") {
      val executor = new SingleThreadOrderExecutor
      val repo = ThreadUnsafeClientRepository(Seq(ClientInfo("1", 0L, 0L, 0L, 0L, 0L), ClientInfo("2", 0L, 0L, 0L, 0L,
        0L)).map(ThreadUnsafeClient.fromClientInfo))
      val future = executor.execute(ReverseOrders(Order(1, "1", OrderType.Buy, "A", 1, 1), Order(1, "2", OrderType.Sell,
        "A", 1, 1)), repo)
      val balancesFuture = for {
        _ <- future
        users <- repo.getAll
        userBalances <- Future.sequence(users.map(u => u.getAllBalances.map(b => u.id -> b)))
      } yield userBalances
      whenReady(balancesFuture, Timeout(Span(10, Seconds))) { case userBalances =>
        val userBalancesMap = userBalances.toMap
        userBalancesMap("1")("A") should be(1)
        userBalancesMap("1")("USD") should be(-1)
        userBalancesMap("2")("A") should be(-1)
        userBalancesMap("2")("USD") should be(1)
      }
    }
    it("should fail on execure orders on unexisted users from repo") {
      val executor = new SingleThreadOrderExecutor
      val repo = ThreadUnsafeClientRepository(Seq(ClientInfo("1", 0L, 0L, 0L, 0L, 0L)).map(ThreadUnsafeClient.fromClientInfo))
      val future = executor.execute(ReverseOrders(Order(1, "1", OrderType.Buy, "A", 1, 1), Order(1, "2", OrderType.Sell,
        "A", 1, 1)), repo)
      future.failed.futureValue shouldBe an[IllegalStateException]
    }
    it("should execure reverse orders of the same user") {
      val executor = new SingleThreadOrderExecutor
      val repo = ThreadUnsafeClientRepository(Seq(ClientInfo("1", 0L, 0L, 0L, 0L, 0L), ClientInfo("1", 0L, 0L, 0L, 0L,
        0L)).map(ThreadUnsafeClient.fromClientInfo))
      val future = executor.execute(ReverseOrders(Order(1, "1", OrderType.Buy, "A", 1, 1), Order(1, "1", OrderType.Sell,
        "A", 1, 1)), repo)
      val balancesFuture = for {
        _ <- future
        users <- repo.getAll
        userBalances <- Future.sequence(users.map(u => u.getAllBalances.map(b => u.id -> b)))
      } yield userBalances
      whenReady(balancesFuture, Timeout(Span(10, Seconds))) { case userBalances =>
        val userBalancesMap = userBalances.toMap
        userBalancesMap("1")("A") should be(0)
        userBalancesMap("1")("USD") should be(0)
      }
    }
  }
} 

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

import com.whisk.docker.impl.spotify.DockerKitSpotify
import org.scalatest._
import org.scalactic._
import com.whisk.docker.scalatest._
import org.scalatest.time.{Second, Seconds, Span}

class SftpSpec
    extends WordSpec
    with Matchers
    with TypeCheckedTripleEquals
    with DockerTestKit
    with DockerKitSpotify
    with DockerSftpService {

  implicit val pc = PatienceConfig(Span(20, Seconds), Span(1, Second))

  "Sftp example" should {
    "use the docker container" in {
      val result = Main.listSftp(SftpConfig(sftpUser, sftpPassword, "localhost", exposedSftpPort))

      result should have length 3
      result.exists(lsEntry => lsEntry.getFilename == sftpDirectory) should ===(true)
    }
  }
} 

//设置package包名称以及导入依赖的类
package im.actor.server

import java.time.Instant

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.time.{ Seconds, Span }
import org.scalatest.{ FlatSpecLike, Inside, Matchers }

import scala.concurrent.ExecutionContext
import im.actor.server.db.DbExtension
import im.actor.server.migrations.v2.{ MigrationNameList, MigrationTsActions }

abstract class BaseAppSuite(_system: ActorSystem = {
                              ActorSpecification.createSystem()
                            })
  extends ActorSuite(_system)
  with FlatSpecLike
  with ScalaFutures
  with MessagingSpecHelpers
  with Matchers
  with Inside
  with ServiceSpecMatchers
  with ServiceSpecHelpers
  with ActorSerializerPrepare {

  protected implicit val materializer: ActorMaterializer = ActorMaterializer()
  implicit lazy val ec: ExecutionContext = _system.dispatcher

  protected implicit lazy val (db, conn) = {
    DbExtension(_system).clean()
    DbExtension(_system).migrate()
    val ext = DbExtension(_system)
    (ext.db, ext.connector)
  }

  system.log.debug("Writing migration timestamps")
  MigrationTsActions.insertTimestamp(
    MigrationNameList.MultiSequence,
    Instant.now.toEpochMilli
  )(conn)
  MigrationTsActions.insertTimestamp(
    MigrationNameList.GroupsV2,
    Instant.now.toEpochMilli
  )(conn)

  override implicit def patienceConfig: PatienceConfig =
    new PatienceConfig(timeout = Span(15, Seconds))

  override protected def beforeAll(): Unit = {
    super.beforeAll()
    db
  }
} 

//设置package包名称以及导入依赖的类
package im.actor.util.http

import java.math.BigInteger
import java.nio.file.Files
import java.security.MessageDigest

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.time.{ Span, Seconds }
import org.scalatest.{ FlatSpec, Matchers }

class DownloadManagerSpec extends FlatSpec with ScalaFutures with Matchers {
  it should "Download https files" in e1

  override implicit def patienceConfig: PatienceConfig =
    new PatienceConfig(timeout = Span(10, Seconds))

  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()

  val downloadManager = new DownloadManager()

  def e1() = {
    whenReady(downloadManager.download("https://ajax.googleapis.com/ajax/libs/webfont/1.5.18/webfont.js")) {
      case (path, size) ?

        val fileBytes = Files.readAllBytes(path)
        fileBytes.length shouldEqual size

        val md = MessageDigest.getInstance("MD5")
        val hexDigest = new BigInteger(1, md.digest(fileBytes)) toString (16)

        hexDigest shouldEqual "593e60ad549e46f8ca9a60755336c7df"
    }
  }
} 

//设置package包名称以及导入依赖的类
package co.s4n.infrastructure.database

import org.cassandraunit.utils.EmbeddedCassandraServerHelper
import org.scalatest.time.{ Millis, Seconds, Span }

import scala.concurrent.duration._

trait CassandraSpec extends TestSuite {

  implicit val defaultPatientConfig =
    PatienceConfig(timeout = Span(15, Seconds), interval = Span(500, Millis))

  override def beforeAll(): Unit = {
    synchronized {
      EmbeddedCassandraServerHelper
        .startEmbeddedCassandra("embedded-cassandra.yaml", 120.seconds.toMillis)
      database.create(5.seconds)
      super.beforeAll()
    }
  }
} 

//设置package包名称以及导入依赖的类
import org.scalatest._
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.time.{Seconds, Span}
import slick.jdbc.PostgresProfile.api._
import slick.jdbc.meta._

class TablesSuite extends FunSuite with BeforeAndAfter with ScalaFutures {
  implicit override val patienceConfig = PatienceConfig(timeout = Span(5, Seconds))

  val suppliers = TableQuery[Suppliers]
  val coffees = TableQuery[Coffees]
  
  var db: Database = _

  def createSchema() =
    db.run((suppliers.schema ++ coffees.schema).create).futureValue
  
  def insertSupplier(): Int =
    db.run(suppliers += (101, "Acme, Inc.", "99 Market Street", "Groundsville", "CA", "95199")).futureValue
  
  before { db = Database.forConfig("h2mem1") }
  
  test("Creating the Schema works") {
    createSchema()
    
    val tables = db.run(MTable.getTables).futureValue

    assert(tables.size == 2)
    assert(tables.count(_.name.name.equalsIgnoreCase("suppliers")) == 1)
    assert(tables.count(_.name.name.equalsIgnoreCase("coffees")) == 1)
  }

  test("Inserting a Supplier works") {
    createSchema()
    
    val insertCount = insertSupplier()
    assert(insertCount == 1)
  }
  
  test("Query Suppliers works") {
    createSchema()
    insertSupplier()
    val results = db.run(suppliers.result).futureValue
    assert(results.size == 1)
    assert(results.head._1 == 101)
  }
  
  after { db.close }
} 

//设置package包名称以及导入依赖的类
package com.knol.db.repo

import org.scalatest.concurrent.ScalaFutures
import com.knol.db.connection.H2DBComponent
import org.scalatest.FunSuite
import org.scalatest.time.Seconds
import org.scalatest.time.Millis
import org.scalatest.time.Span


class BankProductRepositoryTest extends FunSuite with BankProductRepository with H2DBComponent with ScalaFutures {

  implicit val defaultPatience = PatienceConfig(timeout = Span(5, Seconds), interval = Span(500, Millis))

  test("Add new Product ") {
    val response = create(BankProduct("car loan", 1))
    whenReady(response) { productId =>
      assert(productId === 3)
    }
  }

  test("Update bank product ") {
    val response = update(BankProduct("Home Loan", 1, Some(1)))
    whenReady(response) { res =>
      assert(res === 1)
    }
  }

  test("Delete  bank info  ") {
    val response = delete(1)
    whenReady(response) { res =>
      assert(res === 1)
    }
  }

  test("Get product list") {
    val products = getAll()
    whenReady(products) { result =>
      assert(result === List(BankProduct("home loan", 1, Some(1)), BankProduct("eduction loan", 1, Some(2))))
    }
  }

  test("Get bank and their product list") {
    val bankProduct = getBankWithProduct()
    whenReady(bankProduct) { result =>
      assert(result === List((Bank("SBI bank", Some(1)), BankProduct("home loan", 1, Some(1))), (Bank("SBI bank", Some(1)), BankProduct("eduction loan", 1, Some(2)))))
    }
  }

  test("Get all bank and  product list") {
    val bankProduct = getAllBankWithProduct()
    whenReady(bankProduct) { result =>
      assert(result === List((Bank("SBI bank", Some(1)), Some(BankProduct("home loan", 1, Some(1)))), (Bank("SBI bank", Some(1)), Some(BankProduct("eduction loan", 1, Some(2)))), (Bank("PNB bank", Some(2)), None)))
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.knol.db.repo

import org.scalatest.FunSuite
import com.knol.db.connection.H2DBComponent
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.time.{ Millis, Seconds, Span }


class BankRepositoryTest extends FunSuite with BankRepository with H2DBComponent with ScalaFutures {

  implicit val defaultPatience = PatienceConfig(timeout = Span(5, Seconds), interval = Span(500, Millis))

  test("Add new bank ") {
    val response = create(Bank("ICICI bank"))
    whenReady(response) { bankId =>
      assert(bankId === 3)
    }
  }

  test("Update  SBI bank  ") {
    val response = update(Bank("SBI Bank", Some(1)))
    whenReady(response) { res =>
      assert(res === 1)
    }
  }

  test("Delete SBI bank  ") {
    val response = delete(2)
    whenReady(response) { res =>
      assert(res === 1)
    }
  }

  test("Get bank list") {
    val bankList = getAll()
    whenReady(bankList) { result =>
      assert(result === List(Bank("SBI bank", Some(1)), Bank("PNB bank", Some(2))))
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.knol.db.repo

import org.scalatest.concurrent.ScalaFutures
import com.knol.db.connection.H2DBComponent
import org.scalatest.FunSuite
import org.scalatest.time.Seconds
import org.scalatest.time.Millis
import org.scalatest.time.Span


class BankInfoRepositoryTest extends FunSuite with BankInfoRepository with H2DBComponent with ScalaFutures {

  implicit val defaultPatience = PatienceConfig(timeout = Span(5, Seconds), interval = Span(500, Millis))

  test("Add new bank info") {
    val response = create(BankInfo("Goverment", 1000, 1))
    whenReady(response) { bankInfoId =>
      assert(bankInfoId === 2)
    }
  }

  test("Update  bank info ") {
    val response = update(BankInfo("goverment", 18989, 1, Some(1)))
    whenReady(response) { res =>
      assert(res === 1)
    }
  }

  test("Delete  bank info  ") {
    val response = delete(1)
    whenReady(response) { res =>
      assert(res === 1)
    }
  }

  test("Get bank info list") {
    val bankInfo = getAll()
    whenReady(bankInfo) { result =>
      assert(result === List(BankInfo("goverment", 10000, 1, Some(1))))
    }
  }

  test("Get bank and their info list") {
    val bankInfo = getBankWithInfo()
    whenReady(bankInfo) { result =>
      assert(result === List((Bank("SBI bank", Some(1)), BankInfo("goverment", 10000, 1, Some(1)))))
    }
  }

  test("Get all bank and  info list") {
    val bankInfo = getAllBankWithInfo()
    whenReady(bankInfo) { result =>
      assert(result === List((Bank("SBI bank", Some(1)), Some(BankInfo("goverment", 10000, 1, Some(1)))), (Bank("PNB bank", Some(2)), None)))
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.reactivecore.quotes.workshop.api.feature

import com.reactivecore.quotes.workshop.api.QuotesWorkshopApiService
import org.scalatest.concurrent.{PatienceConfiguration, ScalaFutures}
import org.scalatest.time.{Seconds, Span}
import org.scalatest.{FeatureSpec, GivenWhenThen, Matchers}
import play.api.http.Status

class HealthCheckFeature extends FeatureSpec with GivenWhenThen with Matchers with ScalaFutures with PatienceConfiguration {

  override implicit val patienceConfig = PatienceConfig(
    timeout = scaled(Span(1, Seconds))
  )

  feature("health check") {
    scenario("service is healthy") {
      Given("a service")
      val service = QuotesWorkshopApiService()

      When("getting the /health endpoint")
      val response = get(s"http://localhost:${service.port}/health").futureValue

      Then("the response status should be OK")
      response.status should be(Status.OK)

      service.stop()
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.reactivecore.quotes.workshop.api.feature

import com.reactivecore.quotes.workshop.api.QuotesWorkshopApiService
import org.scalatest.concurrent.{PatienceConfiguration, ScalaFutures}
import org.scalatest.time.{Seconds, Span}
import org.scalatest.{FeatureSpec, GivenWhenThen, Matchers}
import play.api.http.Status

class OptionsFeature extends FeatureSpec with GivenWhenThen with Matchers with ScalaFutures with PatienceConfiguration {

  override implicit val patienceConfig = PatienceConfig(
    timeout = scaled(Span(1, Seconds))
  )

  feature("options") {
    scenario("requests options of /items") {
      Given("a service")
      val service = QuotesWorkshopApiService()

      When("requesting for the options of /items endpoint")
      val response = options(s"http://localhost:${service.port}/items").futureValue

      Then("the response status should be OK")
      response.status should be(Status.OK)

      service.stop()
    }
  }

} 

//设置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 com.octo.nad.handson.spark.batch

import com.datastax.spark.connector.cql.CassandraConnector
import com.octo.nad.handson.spark.BatchPipeline
import com.octo.nad.handson.spark.mapping.Revenue
import com.octo.nad.handson.spark.specs.SparkBatchSpec
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.scalatest.time.{Seconds, Span}

class PersistCumulCASpec extends SparkBatchSpec {

  "La méthode persist de l'object Pipeline" should "enregistrer les tuples dans Cassandra" in {


    prepareCassandraKeySpaceAndTables(sc.getConf)
    BatchPipeline.persist(generateRDD)
    
    eventually{

      import com.datastax.spark.connector._
      val result = sc
      .cassandraTable[Revenue](CassandraKeySpace, CassandraCumulTable)
      .collect()

      result.length should be(2)

      result should contain(Revenue("apero", 23L, BigDecimal(1.12)))
      result should contain(Revenue("nettoyage", 10L, BigDecimal(3.40)))
    }(PatienceConfig(scaled(Span(10, Seconds)), Span(1, Seconds)))
  }

  private def generateRDD: RDD[Revenue] = {
    sc.makeRDD(Revenue("apero", 23L, BigDecimal(1.12)) :: Revenue("nettoyage", 10L, BigDecimal(3.40)) :: Nil)
  }

  private def prepareCassandraKeySpaceAndTables(confCassandra: SparkConf) = {
    CassandraConnector(confCassandra).withSessionDo { session =>
      session.execute(s"CREATE KEYSPACE IF NOT EXISTS $CassandraKeySpace WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': $CassandraReplicationFactor }")
      session.execute(s"CREATE TABLE IF NOT EXISTS $CassandraKeySpace.$CassandraCumulTable ($CassandraColumnSection text, $CassandraColumnBatchId bigint, $CassandraColumnRevenue decimal, primary key ($CassandraColumnSection, $CassandraColumnBatchId))")
      session.execute(s"truncate $CassandraKeySpace.$CassandraCumulTable")
    }
  }
}