# `DdDataStreams.Testing`
[🔗](https://github.com/rockneurotiko/dd_data_streams/blob/main/lib/dd_data_streams/testing.ex#L1)

Testing support for a host application.

There are two ways to record with this library in a test, and the first one
is the default:

  * Owner-scoped. Configure `DdDataStreams.Testing.Backend`, start it from
    `test_helper.exs`, and call `start/1` in a test that wants checkpoints.
    Recording is per test, so tests can run with `async: true`, and nothing
    has to be flushed or waited for. This is the API to reach for.
  * Global. `enable!/1` and `capture/1` flip the application environment,
    the shared aggregator and the VM-wide telemetry handler id. They are the
    `async: false` tools for tests that need the full wire-format path
    through the aggregator and the capture transport.

## Owner-scoped tests

    # config/test.exs
    config :dd_data_streams, backend: DdDataStreams.Testing.Backend

    # test/test_helper.exs
    {:ok, _} = DdDataStreams.Testing.Backend.start_link()

    defmodule MyApp.OrdersTest do
      use ExUnit.Case, async: true
      use DdDataStreams.Testing

      test "the consume checkpoint links to the produce checkpoint" do
        DdDataStreams.Testing.start()

        DdDataStreams.consume(%{
          data: payload,
          metadata: %{
            subscription: "projects/demo/subscriptions/orders-sub",
            attributes: Map.new(DdDataStreams.produce("orders-out", payload))
          }
        })

        produce = DdDataStreams.Testing.assert_produced("orders-out")
        consume = DdDataStreams.Testing.assert_consumed("orders-sub")
        DdDataStreams.Testing.assert_chained(produce, consume)
      end
    end

`use DdDataStreams.Testing` only picks private mode for an async case and
shared mode otherwise. A test opts in by calling `start/1`; a test that never
does sees `produce/3` return `[]`, records nothing, and can run alongside the
others. A process a test did not start, such as a Broadway processor, is
allowed with `allow/2`, by pid or by registered name. `checkpoints/1` reads
the result, and `subscribe/0` turns the same points into messages when a test
prefers `assert_receive`.

## Global tests

`enable!/1` and `capture/1` keep their old meaning. They set `:enabled` and
the transport, start the shared aggregator and attach the shared handler id,
so a test using either runs with `async: false`. Use them when the test needs
what only that path gives: an encoded payload through the real aggregator,
or the wire-format bytes the agent would receive.

# `__using__`
*macro* 

Sets up the owner-scoped mode from the ExUnit context.

Picks private mode when the test module is `async: true` and shared mode
otherwise, so a test module only needs:

    use ExUnit.Case, async: true
    use DdDataStreams.Testing

In private mode each test still opts in with `start/1`. In shared mode every
process records into the test process, so `allow/2` is not needed. A module
whose context already chose a mode can call `set_private/0` or `set_global/1`
itself instead.

# `allow`

```elixir
@spec allow(pid() | atom(), pid() | atom()) :: :ok | {:error, struct()}
```

Allows another process to record into `owner`.

Broadway processor and batcher processes are started by Broadway, not by the
test, so they are not in the test's `$callers` and need this call. Both
arguments accept a pid or a registered name:

    DdDataStreams.Testing.start()
    DdDataStreams.Testing.allow(self(), :"my_pipeline.Broadway.Processor_0")

Returns `:ok`, or `{:error, %NimbleOwnership.Error{}}` when the ownership
server refuses the allowance (for example in shared mode, where every process
already records into the same owner).

# `assert_chained`

```elixir
@spec assert_chained(map(), map()) :: map()
```

Asserts that `consume` continues the pathway `produce` started.

`produce` and `consume` are points returned by `assert_produced/2` and
`assert_consumed/2`, or read from `checkpoints/1`. Returns `consume`.

# `assert_consumed`

```elixir
@spec assert_consumed(String.t(), pid() | atom()) :: map()
```

Asserts that a consume checkpoint was recorded for `subscription`.

Returns the point, like `assert_produced/2`.

# `assert_produced`

```elixir
@spec assert_produced(String.t(), pid() | atom()) :: map()
```

Asserts that a produce checkpoint was recorded for `topic`.

Returns the point, so a test can chain more assertions on it:

    produce = DdDataStreams.Testing.assert_produced("orders-out")
    assert "type:google-pubsub" in produce.edge_tags

# `capture`

```elixir
@spec capture((-> any())) :: [map()]
```

Global, `async: false` mode: runs `fun`, flushes every bucket, and returns
the points the transport captured while it ran.

What it flushes is the one shared aggregator, so a test using it must not run
next to another test that records. For the owner-scoped checkpoints a
pipeline records, use `checkpoints/1` instead.

Attaches a handler of its own under a unique id to
`[:dd_data_streams, :transport, :captured]`, so it sees exactly what is
flushed between the start of the call and the flush it makes itself, and
detaches it again whether `fun` returns or raises. The transport must report
on that event, which `DdDataStreams.Transport.Capture` does - pass it to
`enable!/1` - and the aggregator must be running, or the flush has nothing
to send and the result is `[]`.

The flush happens after `fun` returns, so a checkpoint recorded a moment
earlier is assertable without waiting for the ten-second bucket window, the
way `DdDataStreams.Aggregator.flush(:all)` sends it. Every point of every
payload flushed during the call is returned as one flat list of maps with
`:edge_tags`, `:hash`, `:parent_hash`, the three distributions and
`:timestamp_type`, in the shape `decode_payload/1` gives each point.

# `checkpoints`

```elixir
@spec checkpoints(pid() | atom()) :: [map()]
```

The checkpoints recorded for `owner` (the calling process by default), oldest
first.

Each point is a plain map shaped like the points `decode_payload/1` returns:
`:edge_tags`, `:hash`, `:parent_hash`, `:pathway_latency`, `:edge_latency`,
`:payload_size` and `:timestamp_type`. The latencies and the size are the
integers the checkpoint measured, not the flushed distributions, and
`:timestamp_type` is `"current"` because this is the checkpoint itself rather
than an aggregator bucket.

Reading is synchronous: `record/1` in the backend has already stored the
point by the time the call that produced it returns, so no flush or wait is
needed. A test that never called `start/1` gets `[]`. In shared mode the
shared owner's points are returned.

# `decode_payload`

```elixir
@spec decode_payload(binary()) :: map()
```

Decodes a captured transport body into a plain map.

The body is the gzip-compressed MsgPack payload the agent would have
received, as `[:dd_data_streams, :transport, :captured]` reports it; it is
gunzipped and read back into the same shape the library builds it in: atom
keys in this library's vocabulary at every level.

    iex> payload = %DdDataStreams.Payload{env: "prod", service: "checkout"}
    iex> body = payload |> DdDataStreams.Payload.encode() |> IO.iodata_to_binary() |> :zlib.gzip()
    iex> decoded = DdDataStreams.Testing.decode_payload(body)
    iex> {decoded.env, decoded.service, decoded.stats}
    {"prod", "checkout", []}

A payload carries `:env`, `:service`, `:stats`, `:tracer_version`, `:lang`,
`:version`, `:process_tags` and `:product_mask`; each bucket in `:stats`
carries `:start`, `:duration` and its own `:stats` points, and each point
carries `:edge_tags`, `:hash`, `:parent_hash`, `:pathway_latency`,
`:edge_latency`, `:payload_size` and `:timestamp_type`. The three
distributions stay binaries, exactly as they went on the wire; they are
`DdDataStreams.Sketch` protos, not numbers. A key the payload vocabulary
does not know keeps the name it arrived with, so nothing is dropped.

It reads what `DdDataStreams.Payload` encodes; anything else raises.

# `enable!`

```elixir
@spec enable!(keyword()) :: (-> :ok)
```

Global, `async: false` mode: enables the library for one test and returns the
function that undoes it.

This is the aggregator-and-wire-format path. For tests that only want to see
the checkpoints a pipeline records, prefer the owner-scoped API (`start/1`
and `checkpoints/1`), which leaves the shared application environment and
aggregator alone and can run with `async: true`.

Sets `enabled: true` and any of the options `:transport`, `:service` and
`:env:` that are given, attaches the handlers `DdDataStreams.attach/0`
installs, and starts the aggregator if it is not already running, so
checkpoints recorded during the test are aggregated instead of dropped. Any
other option raises an `ArgumentError`, so a typo does not silently configure
nothing.

The returned zero-arity function stops the aggregator this call started,
detaches the handlers this call attached and restores the application
environment to what it was before the call; anything the call found already
running or already attached it leaves alone. It is safe to call twice, and
the aggregator is linked to the calling process, so a test that never runs
the cleanup still cannot leak one. Pass it to `ExUnit.Callbacks.on_exit/1`:

    on_exit(DdDataStreams.Testing.enable!())

# `set_from_context`

```elixir
@spec set_from_context(map()) :: :ok
```

Picks `set_private/0` for an `async: true` case and `set_global/1` otherwise.

# `set_global`

```elixir
@spec set_global(map()) :: :ok
```

Switches the ownership server to shared mode: every process records into the
calling test process.

Raises when the ExUnit context is `async: true`, because a shared owner would
receive another test's checkpoints.

# `set_private`

```elixir
@spec set_private(map()) :: :ok
```

Switches the ownership server to private mode.

This is the default mode and the one `set_from_context/1` picks for an
`async: true` case.

# `start`

```elixir
@spec start(pid() | atom()) :: :ok
```

Makes `owner` (the calling process by default) an owner of the checkpoints.

After this call, `produce/3`, `produce_batch/3`, `consume/1`, `checkpoint/1`
and the Broadway handlers record into `owner`, and `checkpoints/1` reads what
they recorded. Calling it again is harmless. A `Task` records into the owner
that spawned it without this call, because it inherits `$callers`; any other
process needs `allow/2`.

Returns `:ok`. Raises when the ownership server is not running, because a
test that thinks it is recording and is not is worse than a failed test.

In shared mode the shared owner receives the checkpoints, so `start/1` is not
needed there; calling it with the shared owner is harmless.

# `subscribe`

```elixir
@spec subscribe() :: :ok
```

Subscribes the calling owner to checkpoint messages.

Every point recorded after this call is also sent to the owner as
`{DdDataStreams.Testing, :checkpoint, point}`, with the same map
`checkpoints/1` returns. It is the push alternative for tests that prefer
`assert_receive`; `checkpoints/1` is the default because it cannot lose a
point to a mailbox order.

---

*Consult [api-reference.md](api-reference.md) for complete listing*