Scala ZIO Test如何断言RIO返回值 编写压缩函数单元测试

你不需要手动判断返回结果是异常还是成功字节数组，ZIO Test 提供了专门的效果断言组合子，可以直接对RIO/IO效果做断言，不需要手动运行拆包。

可直接运行的测试用例代码 #

import zio._
import zio.blocking._
import zio.test._
import zio.test.Assertion._
import java.io.{ByteArrayOutputStream, DeflaterOutputStream}
import java.nio.charset.StandardCharsets

object CompresserSpec extends DefaultRunnableSpec {

  def spec = suite("Compresser")(
    test("Compress an empty string") {
      val input = ""
      val expected = Array[Byte](0, 0, 0, -1, -1, 3, 0)
      val output = compress(input.getBytes(StandardCharsets.UTF_8))
      assertZIO(output)(equalTo(expected))
    },
    test("Compress null") {
      val output = compress(null)
      assertZIO(output)(fails(isSubtype[NullPointerException](anything)))
    } 
  ).provideCustomLayer(Blocking.live)
}

关键说明 #

• 成功值断言：用assertZIO（ZIO 1.x早期版本可替换为assertM）直接包裹返回的ZIO效果，equalTo断言会自动对比两个字节数组的实际内容。
• 异常断言：用fails组合子匹配效果的失败通道，搭配isSubtype[目标异常类型]即可断言抛出的异常符合预期，不需要手动做try/catch拆包。
• 依赖注入：直接在suite层面用provideCustomLayer(Blocking.live)全局注入Blocking依赖，不需要每个测试单独写provide逻辑，减少重复代码。

原压缩实现的隐藏问题 #

你当前的Compresser实现有并发安全和结果污染的bug：buffer和stream都是全局共享的val，多次调用compress会复用同一个ByteArrayOutputStream和压缩流，不仅多线程下会有并发写入问题，连续调用压缩方法还会把多次输入的内容写进同一个缓冲区，返回错误的压缩结果。
修正后的实现应该把资源创建放在compress方法内部，保证每次调用都持有独立的流资源：

object Compresser {
  def compress(input: Array[Byte]): RIO[Blocking, Array[Byte]] = {
    val managedResource = for {
      buffer <- ZManaged.fromAutoCloseable(ZIO.succeed(new ByteArrayOutputStream()))
      stream <- ZManaged.fromAutoCloseable(ZIO.succeed(new DeflaterOutputStream(buffer, false)))
    } yield (buffer, stream)

    managedResource.use { case (buffer, stream) =>
      for {
        () <- effectBlocking(stream.write(input))
        () <- effectBlocking(stream.flush())
        result = buffer.toByteArray
      } yield result
    }
  }
}

内容的提问来源于stack exchange，提问作者Cypherius