1

我想用 SCIO JobTest 和 Scala Test 进行参数化测试。我使用 TableDrivenPropertyChecks，它允许通过 aa forAll 进行参数化测试。

import org.scalatest.prop.TableDrivenPropertyChecks.{forAll => forAllParams, _}

 val jobArgs = Array(
        "--nullableCoders=true",
        "--inputSubscription=in",
        "--inputAvro=test",
        "--outputBq1=out-table-1",
        "--outputBq2=out-table-2"
      )
    }

val ioParams =
    Table(
      ("description", "inputRawPlusData", "expectedDigitalAvatars", "expectedDataRecords"),
      (
        "Desc1",
        getInputData1...,
        getExpectedOutput1...,
        getExpectedOutput2...
      ),
      (
        "Desc1",
        getInputData2...,
        getExpectedOutput...,
        getExpectedOutput...
      )
    )

forAllParams(ioParams) { (description: String,
                            inputData: Seq[String],
                            expectedOutput1: Seq[...],
                            expectedOutput2: Seq[...]) =>

    it should s"have $description..." in {
      JobTest[com.Job.type]
        .args(jobArgs: _ *)
        .input(PubsubIO[String]("in"), inputData)
        .input(AvroIO[GenericRecord]("test"), test)
        .output(BigQueryIO[Obj1]("out-table-1"))(result => shouldMatchExpectedOutput1(result, expectedOutput1))
        .output(BigQueryIO[Obj2]("out-table-2"))(result => shouldMatchExpectedOutput2(result, expectedOutput2))
        .run()
    }
  }

对于第一组参数，测试工作正常，但对于第二组，Job args 变为空。我不明白为什么（也许是一个状态..）。

当我单独执行所有组参数时（单独和没有其他人），它可以工作。

为什么会有这种行为？

是否可以使用 SCIO JobTest 进行参数化测试？

提前感谢您的帮助。

4

2 回答 2

1

我敢打赌 .args 方法会以某种方式重置数组。将 jobArgs 重新定义为 def 而不是 val 有帮助吗？另一个猜测是 JobTest 需要以不同的方式初始化。像新的 JobTest 什么的

于 2019-12-12T19:02:29.587 回答
0

我终于在一个小工作中重现了这个问题。

我同时使用了 contextAndArgs 和 PipelineOptionFactory ：

  val (sc, args) = ContextAndArgs(cmdlineArgs)

  val options = PipelineOptionsFactory
      .fromArgs(cmdlineArgs: _*)
      .withValidation
      .as(classOf[JobOptions])

    options.setStreaming(true)

我删除了选项部分，这有效。再次感谢 ：）

于 2019-12-13T00:57:14.477 回答