Controllable test services that replace ZIO's standard environment for deterministic testing.
Controllable clock service for testing time-dependent operations.
/**
* Test clock service that provides deterministic time control
*/
trait TestClock extends Clock {
/** Advances the clock by specified duration */
def adjust(duration: Duration): UIO[Unit]
/** Sets clock to specific date and time */
def setDateTime(dateTime: OffsetDateTime): UIO[Unit]
/** Sets clock to specific time offset */
def setTime(duration: Duration): UIO[Unit]
/** Sets time zone for the clock */
def setTimeZone(zone: ZoneId): UIO[Unit]
/** Gets list of pending sleep operations */
def sleeps: UIO[List[Duration]]
/** Gets current time zone */
def timeZone: UIO[ZoneId]
}
object TestClock {
/** Access TestClock service from environment */
def adjust(duration: Duration): URIO[TestClock, Unit]
def setDateTime(dateTime: OffsetDateTime): URIO[TestClock, Unit]
def setTime(duration: Duration): URIO[TestClock, Unit]
def sleeps: URIO[TestClock, List[Duration]]
}Usage Examples:
import zio.test.environment.TestClock
import zio.duration._
testM("time-based operation") {
for {
fiber <- someTTimedOperation.fork
_ <- TestClock.adjust(1.hour)
result <- fiber.join
} yield assert(result)(isSuccess)
}
testM("timeout behavior") {
for {
_ <- TestClock.setTime(0.seconds)
fiber <- longRunningTask.timeout(30.seconds).fork
_ <- TestClock.adjust(31.seconds)
result <- fiber.join
} yield assert(result)(isNone)
}Controllable console service for testing input/output operations.
/**
* Test console service that captures output and provides input
*/
trait TestConsole extends Console {
/** Clears the input buffer */
def clearInput: UIO[Unit]
/** Clears the output buffer */
def clearOutput: UIO[Unit]
/** Clears the error output buffer */
def clearOutputErr: UIO[Unit]
/** Adds input lines to be read */
def feedLines(lines: String*): UIO[Unit]
/** Gets all output lines written */
def output: UIO[Vector[String]]
/** Gets all error output lines written */
def outputErr: UIO[Vector[String]]
}
object TestConsole {
/** Access TestConsole service from environment */
def clearInput: URIO[TestConsole, Unit]
def clearOutput: URIO[TestConsole, Unit]
def feedLines(lines: String*): URIO[TestConsole, Unit]
def output: URIO[TestConsole, Vector[String]]
def outputErr: URIO[TestConsole, Vector[String]]
}Usage Examples:
import zio.test.environment.TestConsole
import zio.console._
testM("console interaction") {
for {
_ <- TestConsole.feedLines("user input", "more input")
_ <- myInteractiveProgram
output <- TestConsole.output
} yield assert(output)(contains("Welcome!"))
}
testM("error output") {
for {
_ <- programThatPrintsErrors
errs <- TestConsole.outputErr
} yield assert(errs)(isNonEmpty)
}Controllable random service for deterministic testing.
/**
* Test random service that provides controllable randomness
*/
trait TestRandom extends Random {
/** Clears boolean values buffer */
def clearBooleans: UIO[Unit]
/** Clears byte array buffer */
def clearBytes: UIO[Unit]
/** Clears character values buffer */
def clearChars: UIO[Unit]
/** Clears double values buffer */
def clearDoubles: UIO[Unit]
/** Clears float values buffer */
def clearFloats: UIO[Unit]
/** Clears integer values buffer */
def clearInts: UIO[Unit]
/** Clears long values buffer */
def clearLongs: UIO[Unit]
/** Clears string values buffer */
def clearStrings: UIO[Unit]
/** Adds boolean values to be returned */
def feedBooleans(booleans: Boolean*): UIO[Unit]
/** Adds byte arrays to be returned */
def feedBytes(bytes: Chunk[Byte]*): UIO[Unit]
/** Adds character values to be returned */
def feedChars(chars: Char*): UIO[Unit]
/** Adds double values to be returned */
def feedDoubles(doubles: Double*): UIO[Unit]
/** Adds float values to be returned */
def feedFloats(floats: Float*): UIO[Unit]
/** Adds integer values to be returned */
def feedInts(ints: Int*): UIO[Unit]
/** Adds long values to be returned */
def feedLongs(longs: Long*): UIO[Unit]
/** Adds string values to be returned */
def feedStrings(strings: String*): UIO[Unit]
/** Sets the random seed */
def setSeed(seed: Long): UIO[Unit]
}
object TestRandom {
/** Access TestRandom service from environment */
def feedInts(ints: Int*): URIO[TestRandom, Unit]
def feedDoubles(doubles: Double*): URIO[TestRandom, Unit]
def feedBooleans(booleans: Boolean*): URIO[TestRandom, Unit]
def setSeed(seed: Long): URIO[TestRandom, Unit]
}Usage Examples:
import zio.test.environment.TestRandom
import zio.random._
testM("deterministic randomness") {
for {
_ <- TestRandom.feedInts(1, 2, 3, 4, 5)
result <- randomGameSimulation
} yield assert(result.winner)(equalTo("Player1"))
}
testM("seeded randomness") {
for {
_ <- TestRandom.setSeed(12345L)
result1 <- randomOperation
_ <- TestRandom.setSeed(12345L)
result2 <- randomOperation
} yield assert(result1)(equalTo(result2))
}Controllable system service for testing environment variables and system properties.
/**
* Test system service that provides controllable system state
*/
trait TestSystem extends System {
/** Clears all environment variables */
def clearEnv: UIO[Unit]
/** Clears all system properties */
def clearProperties: UIO[Unit]
/** Sets environment variable */
def putEnv(name: String, value: String): UIO[Unit]
/** Sets system property */
def putProperty(name: String, value: String): UIO[Unit]
/** Sets line separator */
def setLineSeparator(lineSeparator: String): UIO[Unit]
}
object TestSystem {
/** Access TestSystem service from environment */
def clearEnv: URIO[TestSystem, Unit]
def putEnv(name: String, value: String): URIO[TestSystem, Unit]
def putProperty(name: String, value: String): URIO[TestSystem, Unit]
}Usage Examples:
import zio.test.environment.TestSystem
import zio.system._
testM("environment variables") {
for {
_ <- TestSystem.putEnv("TEST_MODE", "enabled")
_ <- TestSystem.putEnv("API_KEY", "test-key-123")
result <- programThatReadsEnvVars
} yield assert(result)(isSuccess)
}
testM("system properties") {
for {
_ <- TestSystem.putProperty("user.name", "testuser")
result <- programThatReadsSystemProps
} yield assert(result.username)(equalTo("testuser"))
}Complete test environment combining all test services.
/**
* Complete test environment with all controllable services
*/
type TestEnvironment = TestClock with TestConsole with TestRandom with TestSystem
/**
* Complete ZIO test environment
*/
type ZTestEnv = TestClock with TestConsole with TestRandom with TestSystem
object TestEnvironment {
/** Live layer providing test environment */
val live: Layer[Nothing, TestEnvironment]
}
/** Default test environment layer */
val testEnvironment: Layer[Nothing, TestEnvironment]
/** Managed test environment resource */
val testEnvironmentManaged: TaskManaged[TestEnvironment]Service objects for accessing test environment components.
object Annotations {
trait Service {
def annotate[V](key: TestAnnotation[V], value: V): UIO[Unit]
def get[V](key: TestAnnotation[V]): UIO[V]
def supervisedFibers: UIO[SortedSet[Fiber.Runtime[Any, Any]]]
}
def annotate[V](key: TestAnnotation[V], value: V): URIO[Annotations, Unit]
def get[V](key: TestAnnotation[V]): URIO[Annotations, V]
val live: Layer[Nothing, Annotations]
}
object Sized {
trait Service {
def size: UIO[Int]
def withSize[R, E, A](size: Int)(zio: ZIO[R, E, A]): ZIO[R, E, A]
}
def live(size: Int): Layer[Nothing, Sized]
def size: URIO[Sized, Int]
def withSize[R <: Sized, E, A](size: Int)(zio: ZIO[R, E, A]): ZIO[R, E, A]
}
object TestConfig {
trait Service {
def repeats: Int
def retries: Int
def samples: Int
def shrinks: Int
}
def live(repeats: Int, retries: Int, samples: Int, shrinks: Int): ZLayer[Any, Nothing, TestConfig]
val repeats: URIO[TestConfig, Int]
val retries: URIO[TestConfig, Int]
val samples: URIO[TestConfig, Int]
val shrinks: URIO[TestConfig, Int]
}import zio.test._
import zio.test.environment._
import zio.duration._
object TestEnvironmentExample extends DefaultRunnableSpec {
def spec = suite("Test Environment Example")(
testM("clock control") {
for {
start <- clock.nanoTime
_ <- TestClock.setTime(1.hour)
end <- clock.nanoTime
diff = end - start
} yield assert(diff)(isGreaterThan(3600000000000L)) // 1 hour in nanos
},
testM("console interaction") {
for {
_ <- TestConsole.feedLines("Alice", "25")
_ <- putStrLn("Enter name:")
name <- getStrLn
_ <- putStrLn("Enter age:")
age <- getStrLn
output <- TestConsole.output
} yield assert(output)(contains("Enter name:")) &&
assert(name)(equalTo("Alice")) &&
assert(age)(equalTo("25"))
},
testM("deterministic random") {
for {
_ <- TestRandom.feedInts(42, 17, 99)
int1 <- nextInt
int2 <- nextInt
int3 <- nextInt
} yield assert(int1)(equalTo(42)) &&
assert(int2)(equalTo(17)) &&
assert(int3)(equalTo(99))
},
testM("system environment") {
for {
_ <- TestSystem.putEnv("HOME", "/test/home")
_ <- TestSystem.putProperty("user.name", "testuser")
home <- system.env("HOME")
user <- system.property("user.name")
} yield assert(home)(isSome(equalTo("/test/home"))) &&
assert(user)(isSome(equalTo("testuser")))
}
)
}