make specs2 work

manuzhang · manuzhang · commit c5fcb847433d · 2014-10-22T11:34:41.000+08:00
diff --git a/examples/src/main/scala/org/apache/gearpump/streaming/examples/kafka/KafkaBolt.scala b/examples/src/main/scala/org/apache/gearpump/streaming/examples/kafka/KafkaBolt.scala
@@ -69,7 +69,7 @@ class KafkaBolt(conf: Configs) extends TaskActor(conf) {
 
   private def reportThroughput : Unit = {
     val current = System.currentTimeMillis()
-    LOG.info(s"Task $taskId Throughput: ${((count - lastCount), ((current - lastTime) / 1000))} (messages, second)")
+    LOG.info(s"Task $taskId; Actor ${self}; Throughput: ${((count - lastCount), ((current - lastTime) / 1000))} (messages, second)")
     lastCount = count
     lastTime = current
   }
diff --git a/project/Build.scala b/project/Build.scala
@@ -29,6 +29,7 @@ object Build extends sbt.Build {
   val spraySwaggerVersion = "0.4.3"
   val swaggerUiVersion = "2.0.21"
   val specs2Version = "1.13"
+  val mockitoVersion = "1.9.5"
 
   val commonSettings = Defaults.defaultSettings ++ packAutoSettings ++
     Seq(
@@ -97,7 +98,8 @@ object Build extends sbt.Build {
       Seq(
         libraryDependencies ++= Seq(
           "org.apache.kafka" %% "kafka" % kafkaVersion,
-          "org.specs2" %% "specs2" % specs2Version % "test"
+          "org.specs2" %% "specs2" % specs2Version % "test",
+          "org.mockito" % "mockito-core"  % mockitoVersion % "test"
         )
       )
   )  dependsOn(core)
diff --git a/streaming/src/test/scala/org/apache/gearpump/streaming/transaction/OffsetManagerSpec.scala b/streaming/src/test/scala/org/apache/gearpump/streaming/transaction/OffsetManagerSpec.scala
@@ -19,30 +19,33 @@
 package org.apache.gearpump.streaming.transaction
 
 import org.apache.gearpump.streaming.transaction.api.{OffsetManager, Checkpoint, Source}
-import org.apache.gearpump.streaming.transaction.kafka.{KafkaCheckpointManagerFactory, KafkaCheckpointManager, KafkaSource}
+import org.apache.gearpump.streaming.transaction.kafka._
 import org.apache.gearpump.streaming.transaction.kafka.KafkaConfig._
 import org.apache.gearpump.TimeStamp
 import org.apache.gearpump.util.Configs
 import org.specs2.mutable._
 import org.specs2.mock._
 
 
-object OffsetManagerSpec extends Specification with Mockito with MockitoMocker {
+class OffsetManagerSpec extends Specification with Mockito {
   "OffsetManager" should {
     "checkpoint updated timestamp and offsets for each source" in {
 
       "Testing OffsetManager".txt
 
-      val kafkaConfig = mock[Map[String, _]]
       val checkpointManagerFactory = mock[KafkaCheckpointManagerFactory]
       val checkpointManager = mock[KafkaCheckpointManager]
+      val filter = mock[RelaxedTimeFilter]
+      val config = Map(
+        CHECKPOINT_MANAGER_FACTORY_CLASS -> checkpointManagerFactory,
+        CHECKPOINT_FILTER_CLASS -> filter
+      )
 
-      kafkaConfig.getCheckpointManagerFactory returns checkpointManagerFactory
       checkpointManagerFactory.getCheckpointManager(any[Configs]) returns checkpointManager
-      checkpointManager.writeCheckpoint(any[Source], any[Checkpoint]) returns doNothing
+      checkpointManager.writeCheckpoint(any[Source], any[Checkpoint]) answers {args => }
 
 
-      val conf = Configs(kafkaConfig)
+      val conf = Configs(config)
       val offsetManager = new OffsetManager(conf)
 
       val offsetsByTimeAndSource: Map[(Source, TimeStamp), Long] = Map(

Original file line number	Diff line number	Diff line change
`@@ -69,7 +69,7 @@ class KafkaBolt(conf: Configs) extends TaskActor(conf) {`
`69`	`69`
`70`	`70`	`private def reportThroughput : Unit = {`
`71`	`71`	`val current = System.currentTimeMillis()`
`72`		`- LOG.info(s"Task $taskId Throughput: ${((count - lastCount), ((current - lastTime) / 1000))} (messages, second)")`
	`72`	`+ LOG.info(s"Task $taskId; Actor ${self}; Throughput: ${((count - lastCount), ((current - lastTime) / 1000))} (messages, second)")`
`73`	`73`	`lastCount = count`
`74`	`74`	`lastTime = current`
`75`	`75`	`}`