Self-documenting tests

Even though it's 2020, self-documenting tests are still niche. There is Clairvoyance, a Scala flavour of YatSpec (that I mention exensively in this article for IBM). Its creator, Rhys Keepence is an old colleague and told me recently that it "is mostly up to date (although I haven’t yet published for Scala 2.13). The docs on my github page are not super up to date, but the latest version is 1.0.129".

However, introducing yet another new library to the codebase was too much an ask so I started using GivenWhenThen in ScalaTest. It was somewhat painful to get it to print just the Given, When, Then outputs to a separate file that can be version controlled for the edification of other data scientists modulo all the gubbins that are also spat out in your typical build.

I eventually did it using these resources from the ScalaTest docs. The top-and-bottom of it is that the GWT outputs are captured in a Reporter that may be bespoke (IntelliJ uses its own to seperate the GWTs from the logging). This Reporter can then spew out the events at the end of the test. But if you want them in a file produced by your build, you'll need something like this (in Maven):

    <build>

        <plugins>

            <plugin>

                <groupId>org.scalatest</groupId>
                <artifactId>scalatest-maven-plugin</artifactId>
                <version>2.0.0</version>
                <configuration>
                    <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
                    <junitxml>.</junitxml>
                    <stderr/>
                    <filereports>W ../docs/src/main/acceptance_tests/scenarios.txt</filereports>
                </configuration>

...

and then copy scenarios.txt where it can be versioned controlled with something like:

            <plugin>
                <artifactId>maven-resources-plugin</artifactId>
                <version>3.1.0</version>
                <executions>
                    <execution>
                        <id>copy-resources</id>
                        <phase>package</phase>
                        <goals>
                            <goal>copy-resources</goal>
                        </goals>
                        <configuration>
                            <outputDirectory>${basedir}/../docs/src/main/acceptance_tests/</outputDirectory>
                            <resources>
                                <resource>
                                    <directory>target/docs/src/main/acceptance_tests/</directory>
                                    <filtering>true</filtering>

...

The W in filereports means without colour since although this looks good on a Unix CLI, it just adds odd escape characters to a text file, which is what the data scientists want.

I'm unaware of a similar BDD framework for ZIO which means I need to mix my ZIO tests with ScalaTest. Unfortunately, I noticed that with Maven, some ZIO tests were failing but this did not stop the build. I documented this on ZIO's github here.

Encryption

A few random notes I've been making about security libraries I've been using this past year or so.

How Random is Random?

SecureRandom is the gold standard. However, "depending on the implementation, the generateSeed and nextBytes methods may block as entropy is being gathered, for example, if they need to read from /dev/random on various Unix-like operating systems." [JavaDocs] This hasn't been a problem for me so far as I create one million 64-bit random numbers in my unit tests and the whole process takes about a second or two.

On Linux, you can see the temperature of the CPU, fan speeds etc by installing the tools mentioned here (AskUbuntu). This is one way to generate randomness.

There's an interesting addition to the Java API called ThreadLocalRandom that is more efficient than java.util.Random but still not appropriate for secure random number generators.

PGP or GPG?

"OpenPGP is the IETF-approved standard that defines encryption technology that uses processes that are interoperable with PGP. pgp is Symantec's proprietary encryption solution. pgp adheres to the OpenPGP standard and provides an interface that allows users to easily encrypt their files." [NetworkWorld]

"gpg is the OpenPGP part of the GNU Privacy Guard (GnuPG). It is a tool to provide digital encryption and signing services using the OpenPGP standard. gpg features complete key management and all the bells and whistles you would expect from a full OpenPGP implementation." [gpg man pages].

You can have the public key embedded in the file which can identify the recipient.  Why this is useful? "As far as I know, the recipient's public key IDs, key Validity dates, name, and email address are embedded in the GPG ASCII Armor file (GnuPG Manual ). So using pub key file / Key ID / Name / Email to identify which public key to use should all be equivalent." [StackExchange]

Importing a private key

If you haven't got the key, you can't decrypt a file. But if you have, you don't need to specify it. For instance, if I try to decrypt a file for which I don't have a key, I see:

$ gpg --output file.zip -d file.zip.pgp 
gpg: encrypted with RSA key, ID EAC258F9825D4C9C
gpg: decryption failed: No secret key

However, I can import it:

$ gpg --import ~/Temp/key.txt
gpg: key EAC258F9825D4C9C: public key "XXXX-TEST " imported
gpg: key EAC258F9825D4C9C: secret key imported
gpg: Total number processed: 1
gpg:               imported: 1
gpg:       secret keys read: 1
gpg:   secret keys imported: 1

and now decrypt it:

$ gpg --output file.zip -d file.zip.pgp 
gpg: encrypted with 4096-bit RSA key, ID EAC258F9825D4C9C, created 2020-02-27
      "XXXX-TEST "

Bouncy Castle

Bouncy Castle is the defacto library to allow the JVM to access OpenPGP files. 

One gotcha I found when using Bouncy Castle in an über JAR that was called in a Docker container was:

Caused by: java.util.jar.JarException: file:/home/henryp/main-1.0-SNAPSHOT-jar-with-dependencies.jar has unsigned entries ...

There doesn't seem to be a huge amount you can do about this if you insist on using über JARs as "You can't bundle a cryptographic library. They have to be signed for the JVM to load them, and the signature is destroyed when merged into the shadow jar." [GitHub]

This seems to be something specific to Oracle's JDK because if my Docker config file starts with:

FROM openjdk:11-jdk-slim

I don't have this problem. 


Encrypted ZIPs

I was hoping to stream a zip file that was encrypted, decrypting and unzipping as I went but was worried about the ZIP format. Note that a "directory is placed at the end of a ZIP file. This identifies what files are in the ZIP and identifies where in the ZIP that file is located. This allows ZIP readers to load the list of files without reading the entire ZIP archive. ZIP archives can also include extra data that is not related to the ZIP archive." [Wikipedia] 

So, could I really decrypt and unzip a stream?

Changing to GZIP wouldn't help either because "Both zip and gzip use the same compressing format internally, the main difference is in the metadata: zip has it at the end of the file, gzip at the beginning (and gzip only supports one enclosed file easily)." [StackOverflow]

But decrypting the stream and forking a process that unzips it using PipedInputStream and PipedOutputStream seems to work even on files of a about 1gb.

Encrypted Parquet

Parquet Modular Encryption allows certain columns to be encrypted.

OAuth

“Designed specifically to work with … (HTTP), OAuth essentially allows access tokens to be issued to third-party clients by an authorization server, with the approval of the resource owner” [Wikipedia] "A trust store is used to authenticate peers. A key store is used to authenticate yourself." [StackOverflow]

You can get tokens in Google DataFlow with this:

    val credentials = ComputeEngineCredentials.create()
    val accessToken = credentials.refreshAccessToken()
    logger.info(s"accessToken = $accessToken")             // OAuth token
    logger.info(s"getAccount = ${credentials.getAccount}") // service account name

Although almost ubiquitous, OAuth has its drawbacks:

Ross A. Baker @rossabaker Jun 08 04:27

I think the specification is far too complex for what it accomplishes.

I had a lengthy argument implementing it when I worked at a security company as a replacement for a request signing algorithm.

In request signing, you can't just steal a token the way you can in OAuth2.

And the argument is, "Well, it's sent over TLS, what does it matter?"

And as we were having that argument, those tokens were appearing in clear text in our logs.

Was it a shitty implementation? Absolutely. But all it takes is one mistake like that.

It's neither as convenient as basic auth, nor as secure as something like an HMAC-signed request. I feel like it operates in a middle ground that suits no purpose very well.

Gavin Bisesi @Daenyth Jun 08 14:48

another pain point is that IIUC the oauth spec is very full of "MAY" options and relatively few "MUST" options, so every actual implementation does stuff differently and nothing is compatible with anything else eg you often need specific logic to support X vs Y backends

An alternative to OAuth is Request Signing. Basically, the server has all private keys, clients only have their own and messges are encrypted and signed - see Andrew Hoang's blog.

Gotchas

When storing pass phrases etc in files, be careful that your editor does not add a newline. For instance, open a file in vi such:

$ vi /tmp/5Chars.txt

Type the string 12345, save and close it.

$ ls -l /tmp/5Chars.txt

-rw-r--r-- 1 henryp kismet 6 Nov 24 10:23 /tmp/5Chars.txt

What? it's 6 bytes, not 5! One solution is this:

$ echo -n 12345 > /tmp/5Chars.txt 

$ ls -l /tmp/5Chars.txt

-rw-r--r-- 1 henryp kismet 5 Nov 24 10:25 /tmp/5Chars.txt

That's better.

Dependency Injection with ZIO

In a previous post, I showed how to set up a test in ZIO-land. This is more a how it works post.

If you recall, we created a ZIO that needs layers to unit test it. To expand, we have a ZIO that looks like this:

ZIO[Init with Flow with Errors, Throwable, ProblemList]

where Init, Flow and Errors are my bespoke layers and ProblemList is just a type alise for a List of Eithers.

We add these layers later with provideLayer with implementations that are for testing or production. The interesting code has already been called, that is the construction of the ZIO in the first place. For me, it looks like:

  def flow(paths:       Filenames,
           configFile:  String,
           session:     SparkSession,
           fs:          FileSystem): ZIO[Init with Flow with Errors, Throwable, ProblemList] = for {
      s         <- Init.init(configFile, session, fs)
      results   <- Flow.resultsFor(s, paths, session, fs)
...

Now, taking the first element in the for comprehension, we see it looks like:

    def init(configFile:  String,
             session:     SparkSession,
             fs:          FileSystem): ZIO[Init, Throwable, Settings]
      = ZIO.accessM(_.get.initializeWith(configFile, session, fs))

This is where it becomes interesting. 

Although ZIO.accessM looks like we're calling a function, this is syntactic sugar. We're actually receiving a ZIO.AccessMPartiallyApplied[R] that "Effectfully accesses the environment of the effect" [docs].

The _.get returns the service we created (test or production) and initializeWith(...) is just calling my code. But how does get do its magic? Well, Init is just a type alias to Has[A] and "The trait Has[A] is used with ZIO environment to express an effect's dependency on a service of type A" [docs]. How it magics this into existence is down to the Izumi reflect library, whose mysteries I'm only just understanding. 

You can apply any number of layers and leave others dangling. For instance, if we have a:

RIO[KeyVaultLayer with Blocking, A]

(where RIO[R, A] is just an alias for ZIO[R, Throwable, A]) and we add layers so:

val kvService:  ULayer[KeyVaultLayer]               = ...

val zio:        RIO[KeyVaultLayer with Blocking, A] = ...

val partialZio: RIO[Blocking, A]                    = zio.provideSomeLayer[Blocking](kvService)

then you'll notice that partialZio still has one of the original requirements.

Although the ZLayer.fromFunction method seems to allow dependencies between layers, I was scratching my head on how to create a dependency on something that isn't a layer (in my case, a configuration that has been read within the ZIO I want the layer added to). I worked around it by having my layer provide a factory rather than the service itself.  

The take away point of all this is that you can defer adding a dependency until after the for-comprehension code (but before you execute it). This is very convenient for testing.

Cancellation idioms

Java IO interrupt refresher

"The InterruptibleChannel interface is a marker that, when implemented by a channel, indicates that the channel is interruptible... Most, but not all, channels are interruptible.

"Channels introduce some new behaviors related to closing and interrupts. If a channel implements the InterruptibleChannel interface, then it's subject to the following semantics. If a thread is blocked on a channel, and that thread is interrupted (by another thread calling the blocked thread's interrupt() method), the channel will be closed, and the blocked thread will be sent a ClosedByInterruptException.  Additionally, if a thread's interrupt status is set, and that thread attempts to access a channel, the channel will immediately be closed, and the same exception will be thrown." [Java NIO, Hitchens]

Summarising, if a thread is interrupted before or during a blocking call on a channel, the channel is closed and an exception is thrown.

"It may seem rather draconian to shut down a channel just because a thread sleeping on that channel was interrupted. But this is an explicit design decision made by the NIO architects.  Experience has shown that it's impossible to reliably handle interrupted I/O operations consistently across all operating systems."

"Interruptible channels are also asynchronously closable. A channel that implements InterruptibleChannel can be closed at any time, even if another thread is blocked waiting for an I/O to complete on that channel. When a channel is closed, any threads sleeping on that channel will be awakened and receive an AsynchronousCloseException. The channel will then be closed and will be no longer usable." [ibid]

The problems with Java

Daniel Spiewak @djspiewak Apr 24 19:47, 2020

There are a couple things with thread interruption that are horrible:

There's no way to build "uninterruptible" code. Meaning that you cannot have a critical section which acquires a resource atomically in several steps. Or in other words, there is no analogue to the acquire action in bracket.

The only way to detect self-cancelation for valid purposes (e.g. resource cleanup) is catching the InterruptedException, but doing this immediately flips the interrupted bit on the Thread back to false! 

The only solution to this is to do Thread.currentThread().interrupt() at the end of your exception handler, which almost no one knows to do. To make matters more annoying, even if you do this correctly, you mess up the stack trace on the interruption, because it's technically a new interrupt.

Oh, and exception handlers are not critical regions either, so if someone is just hammering the interrupt() button over and over externally, you could catch the exception, try to clean things up, and then get immediately interrupted again. This actually happens a lot because of the next point.

Catching Exception or Error will silently catch InterruptedException, even when that's almost guaranteed to not be what you want to do. This leads to silently ignoring interruption in most code paths, which is why people repeatedly hammer interrupt() in the first place.

The problems also go deeper than just Java but down to the OS level. "The underlying stream may not know it its closed until you attempt to write to it (e.g. if the other end of a socket closes it)" [SO]. "There's no API for determining whether a stream has been closed." [SO]

The semantics of cancelling

This is a complicated area. (see the interruption model proposed for Cats Effects 3 at https://github.com/typelevel/cats-effect/issues/681)

Basically interruptible/uninterruptible are not composable. The best way to think about it is that "interruptable means always accept the interrupt, no matter what", while "uninterruptible means always suppress the interrupt no matter what". But the uninterruptible(fa >> interruptible(fb) >> fc) breaks either one of the guarantees.
So you have to choose: do you want resource leaks (by biasing in favor of the innermost in that context), or do you want possible deadlocks (by biasing in favor of the outermost)?
And you can't even phrase it as inner/outer, because you can do the same thing in reverse: uninterruptible(interruptible(fa >> uninterruptible(fb) >> fc))
..
as prior art here, Haskell tried all of these and ultimately decided that mask/poll was the sanest solution [Daniel Spiewak, Gitter]

An alternative architecture

"it's a mistake to think of interruptibility as being an attribute of threads or fibers. Instead it should be an attribute of the activities which run on the threads/fibers, and of necessity, that means that any interruptible activity must have it's own first class interrupt channel. If we go down that route then an activity is interruptible if it 1) has an interrupt channel and 2) it's interrupt channel is accessible. If it doesn't have an interrupt channel, or the channel is hidden somehow, then it's uninterruptible.

"Exposing an explicit interrupt channel on every blocking operation that we want to be interruptible is obviously a lot more laborious than just firing random interrupts at globally visible threads/fibers and hoping for the best, but I think it's the only way to go."
[Miles Sabin]

How does this affect Effectful Systems?

Integrating Scala code that uses effectful libraries with Java IO code can cause problems.

Gavin Bisesi @Daenyth Feb 05 21:16, 2020

InputStream is always a blocking api

(note you don't need much to make a blocker; Blocker.apply gives a Resource of one)

Daniel Spiewak @djspiewak Feb 05 21:46, 2020

FYI, all things involving files are blocking except on Windows, and even then they're blocking most of the time.

So the "NIO stuff" that is inside of getResourceAsStream is actually not NIO but rather regular IO wrapped up with a thread pool :-(

I generally use Blocker just to be safe on resource access. It doesn't really cost that much in terms of syntax

... non-blocking things must have a callback-driven API

either directly (via callbacks passed to functions) or indirectly (via Future or CompletableFuture)

If something doesn't have a callback API, then you know it's blocking

When can you cancel?

The effect of a cancellation is felt at every asynch boundary or every N flatMaps (where N=1 for ZIO, it seems).

Note that there is no code in ZIO nor Cats that calls Thread.interrupt() that I could find. Note however that ZIO still gives you the ability to wrap your code in a Future and cancelling this will lead to an InterruptedException (see this gist).

Note that there are still undefined areas in Cats regarding cancellation:

Raas Ahsan @RaasAhsan Jul 07 20:26

calling cancel right after start results in non-deterministic behavior

Fabio Labella @SystemFw Jul 08 19:38

yeah I wanted to say

fa.guarantee(foo) doesn't guarantee that foo will always happen

it guarantees that if fa happens, then foo always happen

In particular if you have fa.guarantee(foo).start.flatMap(_.cancel), foo might happen or not, because the program can be cancelled before fa gets scheduled to run

See tip #2 at this Cats video (An Introduction to Interruption by Jakub Kozlowskiat 11'03" ) where starting and joining in a for comprehension is an anti-pattern (what if one fails to complete?). Instead, one should use (ioa, iob).parTupled.

Does this still happen in 2020?

Lazy data modellers kick the can down the road when choosing an inappropriate type. For instance, we had a field that could only be a calendar date modelled as a java.sql.Timestamp and was told to ignore the time element. 

The trouble is, Timestamp contains no timezone information. To illustrate the problem here, a colleague in Romania created a Parque file with Spark that contained today's date at 0 hours, 0 minutes and 0 seconds - midnight, right? He then sent it to me only for me to see:

scala> val df = spark.read.parquet("/home/henryp/Downloads/part-00000-d41a228d-65fe-47ff-a70f-825a3cc61846-c000.snappy.parquet")

df: org.apache.spark.sql.DataFrame = [Timestamp: timestamp]

scala> df.show()

+-------------------+

|          Timestamp|

+-------------------+

|2020-07-12 22:00:00|

+-------------------+

Horrors - that's yesterday! (Romania is currently 2 hours ahead of the UK).

Of course, it should be 2020-07-13 00:00:00, right? Well, no. Since I'm currently in the BST timezone, that's actually 1 hours off UTC and only the godless would use anything but UTC.

The problem is compounded by the string representation of the dates being specific to the locale of the programmer. My Romanian colleage might have double checked his data before sending it and been lulled into a false sense of security that the timestamps indeed had a zeroed time component. 

ZIO cheats

Here are some brief notes I made to help myself become more familiar with ZIO.

The ZIO Monad

The ZIO monad is defined as ZIO[-R, +E, +A] where R is the requirement, the E is the error and the A is the value.

To feed the requirements into ZIO, use the compose function, its >>> synonym or a provideXXX function.

Useful type aliases

A U in ZIO type names indicates that some code cannot error. For instance:

UIO[+A]      = ZIO[Any, Nothing, A]
URIO[-R, +A] = ZIO[R,   Nothing, A]

Mnemonic: think of that U as an indication the monad is Unexceptional.

Note that you can't have a reference to Nothing. For example:

Welcome to the Ammonite Repl 2.0.4 (Scala 2.13.1 Java 1.8.0_241)

@ val x: Nothing = "Hello" 

cmd0.sc:1: type mismatch;

 found   : String("Hello")

 required: Nothing

val x: Nothing = "Hello"

Since something can never be Nothing, we know that branch of the code cannot be executed. 

Error handling

... is perhaps simpler in ZIO when compared to Cats. You simply call catchAll on your ZIO monad (see this article for fallback, folding and retries).

If we want to ignore an exception then you can do something like:

    val result:     ZIO[Any, DBError, Unit] = makeUser.provideLayer(fullLayer)
    val exit:       UIO[Int]                = UIO(1)   
    val resultExit: ZIO[Any, DBError, Int]  = result *> exit

    val afterCatch: ZIO[Any, Nothing, Int]  = resultExit.catchAll(_ => exit)

Obviously, this won't compile:

  override def run(args: List[String]): ZIO[zio.ZEnv, Nothing, Int] = {
    val exitResult: ZIO[Any, DBError, Unit] = exit *> result
    exitResult.catchAll(_ => exit)

  }

as we've only changed the error channel type but the value value is still Unit.

If you want to only take the successful results, you need something like:

  val xs:        List[IO[Throwable, String]] = ???

  val successes: UIO[List[String]]           = ZIO.collectAllSuccesses(xs)

One way to get you hands on an error is using ZIO.flip which can be useful in testing [SO].

Combinators

The joke in the Cats world is that the answer is always Traverse. But in ZIO, this operation is collectAll (or collectAllPar if you want the work done in parrallel). It's a pity that that ZIO didn't call it Sequence (a close relative of Traverse) as this has precedent in the plain old Scala world in Future.sequence.

jdegoes 12 April 2020 at 2:10 PM (Discord)

x1.ignore &> x2.ignore &> x3. This will execute x1, x2, and x3 in parallel, using zipRightPar  (that's the &> operator, zips two effects together in parallel, returning whatever is produced on the right), and ignore the result of x1 and x2 so their failures don't influence the result of the computation.

 

In ZIO, even parallel zip or collect or foreach operations will "kill" the other running effects if one of them fails. Because that's often what you want. To get the other behavior, just use .ignore in the right places to ignore the failures you don't care about.

Threads

ZIO effects can be executed on other Scala ExecutionContexts with the use of ZIO.on.

Alternatively, you might like to use zio.blocking.XXX which "provides access to a thread pool that can be used for performing blocking operations, such as thread sleeps, synchronous socket/file reads, and so forth. The contract is that the thread pool will accept unlimited tasks (up to the available memory) and continuously create new threads as necessary." [JavaDoc]

For the purposes of cancellation: 

adamfraser 25/6/2020 at 11:13 PM (Discord)

Interruption is checked between each flatMap.

If for whatever reason you can't run your code in a zio.App, you can call zio.Runtime.default.unsafeRun on your effect.

Interop with plain Scala

ZIO maps nicely to plain Scala classes. For instance, ZIO[?, E, A].either yields a ZIO that wraps a plain, old Scala Either[E, A]. Going the other way is simply a ZIO.fromEither call.

Similarly, .option will yield a ZIO that wraps a None or Some[A].

Resource management

ZIO resourcement management is syntactically similar to Cats where bracket takes the code to close the resource and the code that uses the resource:

  val toInputStream: Task[InputStream] = ...

  toInputStream(resource).bracket(close(_)) { input =>

    ZIO {

       new FileOutputStream(output)

    }.bracket(close(_)) { output =>

      ZIO { IOUtils.pipe(input, output) }

    }

Testing

If you want ZIO tests to be picked up as part of a build that uses JUnit then your class should extend zio.test.junit.JUnitRunnableSpec.

You can fail tests on timeout with:

import zio.test.TestAspect._

import zio.duration._

  override def spec: ZSpec[TestEnvironment, Any] = suite("...")(testM("..."){

...

  } @@ timeout(10 seconds)

Unit testing with ZIO

ZIO has the concept of Layers. These allow you to inject requirements to your ZIO monad (the R in ZIO[R, E, A]). The nice thing about them is that they make testing easier.

Say, I have an application with an intialisation piece and a data flow piece. My code would look like this:

import zio.{Has, Task, ZIO}

  type Init = Has[Init.Service]

  object Init {

    trait Service {

      def initializeWith(configFile:  String,

                         session:     SparkSession,

                         fs:          FileSystem): ZIO[Any, Throwable, Settings]

    }

...

  type Flow = Has[Flow.Service]

  object Flow {

    trait Service {

      def flow(s:       Settings,

               paths:   Filenames,

               session: SparkSession,

               fs:      FileSystem): Task[Results[String]]

    }

Now, the data structure I want to test is something like a ZIO[Init with Flow, Throwable, A] and the code that creates it is probably some large and complicated for-comprehension that's begging for a test. 

In production, the code to "run" it would look like:

    val prodInitialization: ULayer[Init]    = ZLayer.succeed(InitProd)

    val prodDataFlow:       ULayer[Flow]    = ZLayer.succeed(FlowProd)

...

    zio.provideLayer(prodInitialization ++ prodDataFlow)

where InitProd and FlowProd are my production objects that implement Init and Flow. 

My test code, however, look like:

import zio.test.Assertion._

import zio.test.environment.TestEnvironment

import zio.test.junit.JUnitRunnableSpec

import zio.test.{ZSpec, assertM, suite, testM}

import zio.{Layer, Task, ZIO, ZLayer}

class InitFlowSpec extends JUnitRunnableSpec {

  def unhappyPathInitLayer(e: Exception): Layer[Nothing, Init] = ZLayer.succeed(

    new Init.Service {

      override def initializeWith(configFile: String, session: SparkSession, fs: FileSystem) = ZIO.fail(e)

    }

  )

...

  val error                         = new Exception("Boo")

  val withNoErrors: Results[String] = Map.empty

  override def spec: ZSpec[TestEnvironment, Any] = suite("Orchestration")(

    testM("returns error if initialisation fails"){

      val layers    = unhappyPathInitLayer(error) ++ flowLayer(withNoErrors)

      val result    = zio.provideLayer(layers)

      assertM(result.run)(fails(equalTo(error)))

    }

  )

Et voila, we have a mocking framework for the requirement channel of a ZIO monad.