Testing

A watch you have never seen fire is a watch you are guessing about. Kepler is built to be tested directly: you can control the clock, control the tick, and receive events in the test process.

The setup

Start a Kepler per test with explicit configuration, a callback sink pointed at the test process, and a tick long enough that it never fires on its own.

defmodule MyApp.WatchesTest do
  use ExUnit.Case, async: false

  setup do
    start_supervised!(
      {Kepler.Supervisor,
       watches: MyApp.Watches,
       tick: 60_000,
       sinks: [test: {Kepler.Sink.Callback, handler: &send(self(), {:kepler, &1})}]}
    )

    :ok
  end
end

Three things are doing work here:

• start_supervised! ties Kepler's lifetime to the test, so each test gets a clean install with fresh counters.
• tick: 60_000 means the only evaluation that happens is the one you ask for with Kepler.tick/0. No sleeping, no polling, no flakiness.
• The callback sink delivers the event to your test process, so assert_receive is all you need.

async: false is required. Kepler's processes are named, so only one runs per node, and telemetry handler attachment is global.

Your application's Kepler is in the way

If :kepler is configured in config/test.exs, the application-started tree already holds the named processes and start_supervised! will fail with :already_started. Stop it once, in test/test_helper.exs:

Application.stop(:kepler)

ExUnit.start()

Alternatively set config :kepler, enabled: false in config/test.exs — but that still occupies the process names, so stopping the application is the cleaner option when you want per-test installs.

Testing a telemetry watch

test "checkout latency fires when p99 crosses the threshold" do
  for _i <- 1..50 do
    :telemetry.execute(
      [:my_app, :checkout, :stop],
      %{duration: System.convert_time_unit(3_000, :millisecond, :native)},
      %{}
    )
  end

  Kepler.tick()

  assert_receive {:kepler, %Kepler.Event{watch: :checkout_latency, state: :firing} = event}
  assert event.context.measurement.value >= 3_000
  assert event.context.measurement.unit == :millisecond
end

Emit through :telemetry.execute/3, exactly as your application does — that exercises the real handler, the real counters, and the real unit conversion. If your code emits the event through a wrapper, call the wrapper.

Assert the negative too. It is the assertion that catches a threshold you got backwards:

test "stays quiet below the threshold" do
  for _i <- 1..50 do
    :telemetry.execute([:my_app, :checkout, :stop], %{duration: 5}, %{})
  end

  Kepler.tick()

  refute_receive {:kepler, %Kepler.Event{watch: :checkout_latency}}, 100
end

Testing a process watch

test "the export backlog fires when the queue builds up" do
  worker = spawn(fn -> Process.sleep(:infinity) end)
  Process.register(worker, MyApp.ExportWorker)
  on_exit(fn -> await_exit(worker) end)

  for _i <- 1..5, do: send(MyApp.ExportWorker, :work)
  Kepler.tick()

  assert_receive {:kepler, %Kepler.Event{watch: :export_backlog}}
end

defp await_exit(pid) do
  ref = Process.monitor(pid)
  Process.exit(pid, :kill)
  assert_receive {:DOWN, ^ref, :process, ^pid, _reason}, 1_000
end

A process that never receives accumulates a mailbox you can measure. Wait for it to actually die in teardown — Process.exit/2 is asynchronous and the registered name is only freed once the process is gone, so the next test will race it otherwise.

Testing sustained:

sustained: is measured against a monotonic clock, so a watch declared with sustained: :timer.seconds(30) genuinely takes thirty seconds of ticks to fire. Do not sleep for it. Test the two halves separately:

Test the state machine directly. Kepler.Trigger is pure, and the clock is an argument:

test "holds for the full window before firing" do
  opts = [sustained: 30_000, cooldown: 0, notify_resolved: false]

  assert {:none, t} = Kepler.Trigger.step(Kepler.Trigger.new(), true, 0, opts)
  assert {:none, t} = Kepler.Trigger.step(t, true, 29_999, opts)
  assert {:fire, _t} = Kepler.Trigger.step(t, true, 30_000, opts)
end

Test that your condition is right by asserting the watch reaches :pending:

test "the condition becomes true" do
  emit_slow_checkouts()
  Kepler.tick()

  latency = Enum.find(Kepler.status().watches, &(&1.name == :checkout_latency))
  assert latency.state == :pending
end

Together those cover the whole behaviour without a clock in the test.

Inspecting rather than receiving

Kepler.status/0 reports every watch's current state and value, which is often a better assertion than an event — it tells you why nothing fired.

test "the watch is seeing data at all" do
  emit_some_events()
  Kepler.tick()

  backlog = Enum.find(Kepler.status().watches, &(&1.name == :export_backlog))

  assert backlog.value != nil, "the source produced no data"
  assert backlog.state == :idle
end

A value of nil means the source produced nothing — usually a process that is not registered under the name you declared, or a telemetry event name that does not match what your application emits. That is the most common reason a watch does not fire, and it is invisible from refute_receive alone.

Waiting for delivery

Sinks run in supervised tasks, so delivery is asynchronous. assert_receive handles that for a callback sink. When you need every delivery finished — for example before asserting on Kepler.status().emitter — use Kepler.drain/1:

Kepler.tick()
assert Kepler.drain() == :ok

assert Kepler.status().emitter.dropped == 0

It returns {:error, :timeout} rather than raising if the queue does not drain, because a wedged sink is a normal thing for Kepler to survive.

Testing a custom sink

Sinks are plain modules with two callbacks, so test them without Kepler running:

test "pages on a severity: :page event" do
  {:ok, state} = MyApp.PagerSink.init(routing_key: "test")

  event =
    Kepler.Event.new(
      Kepler.Watch.new(
        name: :checkout_latency,
        source: {:telemetry, [:my_app, :checkout, :stop]},
        tier: 1,
        measure: %{key: :duration, aggregate: {:percentile, 99}},
        severity: :critical
      ),
      :firing,
      %{value: 2_431, prev: 1_980, delta: 451, rate: 451.0, window_ms: 1_000}
    )

  assert MyApp.PagerSink.deliver(event, state) == :ok
end

If your sink implements format/1, test the two halves separately — that is the point of the split:

payload = MyApp.PagerSink.format(event)
assert payload.severity == "critical"
assert MyApp.PagerSink.deliver(payload, state) == :ok

For the webhook sink, implement a Kepler.Transport that records instead of sending, and pass it as transport:. That covers your headers, signing, and payload without a socket.

Testing a crash watch

Crash a real process. GenServer.start/2 rather than start_link/2, so the crash does not take the test with it:

defmodule CrashingWorker do
  use GenServer

  def start(opts \\ []), do: GenServer.start(__MODULE__, opts)

  @impl true
  def init(_opts) do
    Logger.metadata(request_id: "req-42")
    {:ok, %{pending: 3}}
  end

  @impl true
  def handle_cast(:boom, _state), do: raise("kaboom")
end

test "a crash carries the process's state and the request that caused it" do
  {:ok, pid} = CrashingWorker.start()
  ref = Process.monitor(pid)

  capture_log(fn ->
    GenServer.cast(pid, :boom)
    assert_receive {:DOWN, ^ref, :process, ^pid, _reason}, 2_000
  end)

  assert_receive {:kepler, %Kepler.Event{watch: :process_crash} = event}, 2_000

  assert event.context.enriched.process_state == %{pending: 3}
  assert event.context.enriched.last_message == {:"$gen_cast", :boom}
  assert event.context.enriched.request_context.request_id == "req-42"
  assert %RuntimeError{} = event.context.enriched.reason
end

capture_log/1 keeps the crash report out of your test output. It does not stop Kepler seeing it — Kepler has its own :logger handler.

One crash produces three OTP reports, so assert that it fires once:

assert_receive {:kepler, %Kepler.Event{watch: :process_crash}}, 2_000
refute_receive {:kepler, %Kepler.Event{watch: :process_crash}}, 200

If a supervisor_report: watch never fires in your tests, check config :logger, handle_sasl_reports: true — see crash attribution.

Testing a system monitor watch

Send the monitor message directly rather than trying to provoke the VM into a long garbage collection:

test "a long GC fires" do
  send(Kepler.SystemMonitor, {:monitor, self(), :long_gc, [timeout: 900]})

  assert_receive {:kepler, %Kepler.Event{watch: :long_gc} = event}
  assert event.context.tier == 0
  assert event.context.detail[:timeout] == 900
end

This exercises the real routing, including per-watch threshold matching.

Restore the node's system monitor in teardown — there is only one, and taking it changes global state for the rest of the run:

setup do
  previous = :erlang.system_monitor()
  on_exit(fn -> restore(previous) end)
  :ok
end

Watching Kepler itself

Kepler emits its own telemetry, which is useful in tests and in production:

:telemetry.attach("test", [:kepler, :event, :dropped], fn _e, m, meta, pid ->
  send(pid, {:dropped, m, meta})
end, self())

Compile-time failures need no test

Unknown measurements, unknown identifiers in conditions, duplicate watch names, and cycles between conditions are all build failures. You do not need tests for them — you cannot ship a module containing them.

If you are writing a library that generates Kepler declarations and want to assert on those failures, Kepler.CompileError is the exception, and Code.compile_string/2 is the way to trigger it.