How to Monitor StreamData with Vigilmon

StreamData is Elixir's built-in property-based testing library, integrated with ExUnit. Rather than writing individual example-based tests, you define properties — invariants that must hold across a wide range of inputs — and StreamData generates hundreds or thousands of random inputs to find counterexamples. When a failure is found, it shrinks the input to the smallest failing case.

Property-based tests catch edge cases that hand-crafted examples miss: empty strings, maximum integers, unexpected Unicode, deeply nested structures, and degenerate combinations. But they are only catching those bugs if they actually run. Vigilmon heartbeats paired with StreamData CI runs ensure your property test suite stays green and keeps executing.

Why Monitor StreamData?

Property-based tests are powerful but have unique failure modes beyond ordinary ExUnit tests:

• Generator crashes — a custom generator raises on certain seeds; the property never gets data to test, it just errors immediately
• Shrinking timeout — a complex property takes 60 seconds to shrink a counterexample and CI kills the job; the failure is logged as a timeout, not a property violation
• Flaky randomness — a property that passes 999 times out of 1000 ships to production where it fails; without a heartbeat you don't know when CI last ran all properties
• Configuration drift — max_runs: 100 is set in CI but development uses 1000; a counterexample that needs 500 draws to surface escapes CI
• Generator composition bugs — ExUnitProperties.gen all pipelines with filter/2 can produce generators that exhaust their sample pool; ExUnit reports a StreamData.FilterTooNarrowError
• Property coverage gaps — new code paths are added without corresponding properties; the heartbeat stops confirming that coverage keeps pace

Key Metrics to Track

| Metric | What it tells you | |--------|-------------------| | Property test pass rate | Whether all properties hold across all generated inputs | | Generator error count | Generators raising or exhausting their filter pool | | Shrink depth | How long shrinking takes — deep shrinking may signal performance issues | | max_runs configuration | Whether CI uses the same run count as development | | CI heartbeat | Whether property tests ran recently | | Total draws per run | Whether randomized inputs are actually diverse | | Seed reproducibility | Whether failing seeds can be replayed deterministically |

Step 1: Add StreamData to Your Project

StreamData ships with Elixir's standard test tooling:

# mix.exs
defp deps do
  [
    {:stream_data, "~> 1.1", only: [:test, :dev]},
    # rest of your deps
  ]
end

Import ExUnitProperties in your test files:

# test/test_helper.exs
ExUnit.start()

# test/my_app/property_test.exs
defmodule MyApp.PropertyTest do
  use ExUnit.Case, async: true
  use ExUnitProperties
end

Step 2: Write Properties for Core Invariants

# test/my_app/parser_property_test.exs
defmodule MyApp.ParserPropertyTest do
  use ExUnit.Case, async: true
  use ExUnitProperties

  alias MyApp.Parser

  property "parse and serialize are inverse operations" do
    check all input <- string(:printable, min_length: 1, max_length: 10_000),
              max_runs: 500 do
      parsed = Parser.parse(input)
      serialized = Parser.serialize(parsed)
      re_parsed = Parser.parse(serialized)
      assert re_parsed == parsed,
        "Round-trip failed for input of length #{String.length(input)}"
    end
  end

  property "parse never raises on any binary input" do
    check all input <- binary(min_length: 0, max_length: 100_000),
              max_runs: 1_000 do
      try do
        Parser.parse(input)
        :ok
      rescue
        e ->
          flunk("Parser.parse/1 raised #{Exception.message(e)} on binary input")
      end
    end
  end

  property "empty input returns empty result" do
    check all _ <- constant(nil),
              max_runs: 1 do
      result = Parser.parse("")
      assert result == %{}, "Empty input should return empty map"
    end
  end
end

Step 3: Write Custom Generators

# test/support/generators.ex
defmodule MyApp.Generators do
  use ExUnitProperties

  def user_params do
    gen all first_name <- string(:alphanumeric, min_length: 1, max_length: 50),
            last_name  <- string(:alphanumeric, min_length: 1, max_length: 50),
            age        <- integer(18..120),
            email      <- user_email() do
      %{
        first_name: first_name,
        last_name:  last_name,
        age:        age,
        email:      email,
      }
    end
  end

  def user_email do
    gen all local  <- string(:alphanumeric, min_length: 1, max_length: 30),
            domain <- member_of(["example.com", "test.io", "vigilmon.online"]) do
      "#{local}@#{domain}"
    end
  end

  def non_empty_list(inner_generator) do
    gen all head <- inner_generator,
            tail <- list_of(inner_generator, max_length: 20) do
      [head | tail]
    end
  end
end

Test the generators themselves:

# test/support/generators_test.exs
defmodule MyApp.GeneratorsTest do
  use ExUnit.Case, async: true
  use ExUnitProperties

  alias MyApp.Generators

  property "user_params generator produces valid users" do
    check all params <- Generators.user_params(), max_runs: 100 do
      assert is_binary(params.first_name)
      assert is_binary(params.last_name)
      assert params.age in 18..120
      assert String.contains?(params.email, "@")
    end
  end

  property "user_email generator never produces empty strings" do
    check all email <- Generators.user_email(), max_runs: 200 do
      assert String.length(email) > 0
      assert String.contains?(email, "@")
    end
  end

  property "non_empty_list generator always returns at least one element" do
    check all list <- Generators.non_empty_list(integer()), max_runs: 200 do
      assert length(list) >= 1
    end
  end
end

Step 4: Add a Generator Smoke Test Mix Task

# lib/mix/tasks/check_generators.ex
defmodule Mix.Tasks.CheckGenerators do
  use Mix.Task

  @shortdoc "Validate StreamData generators produce well-formed data"

  def run(_args) do
    generators = [
      {"user_params", fn -> MyApp.Generators.user_params() end},
      {"user_email",  fn -> MyApp.Generators.user_email() end},
    ]

    results =
      Enum.map(generators, fn {name, gen_fn} ->
        try do
          gen = gen_fn.()
          samples = Enum.take(gen, 50)

          if length(samples) < 50 do
            {name, {:too_few_samples, length(samples)}}
          else
            {name, :ok}
          end
        rescue
          e -> {name, {:error, Exception.message(e)}}
        end
      end)

    failed = Enum.filter(results, fn {_name, status} -> status != :ok end)

    if failed == [] do
      Mix.shell().info("All StreamData generators OK (#{length(generators)} checked, 50 samples each)")
    else
      Enum.each(failed, fn {name, status} ->
        Mix.shell().error("FAIL #{name}: #{inspect(status)}")
      end)
      exit({:shutdown, 1})
    end
  end
end

Step 5: Run Property Tests in CI with a Heartbeat

# .github/workflows/property-tests.yml
name: StreamData Property Tests

on:
  push:
    branches: [main]
  schedule:
    - cron: '0 5 * * *'   # daily at 05:00 UTC

jobs:
  property-tests:
    name: StreamData Properties
    runs-on: ubuntu-latest
    env:
      VIGILMON_STREAMDATA_HEARTBEAT_URL: ${{ secrets.VIGILMON_STREAMDATA_HEARTBEAT_URL }}
      STREAMDATA_RUNS: 500
    steps:
      - uses: actions/checkout@v4

      - uses: erlef/setup-beam@v1
        with:
          elixir-version: '1.17'
          otp-version: '27'

      - name: Cache deps
        uses: actions/cache@v3
        with:
          path: deps
          key: ${{ runner.os }}-deps-${{ hashFiles('mix.lock') }}

      - run: mix deps.get

      - name: Validate generators
        run: MIX_ENV=test mix check_generators

      - name: Run property tests
        run: mix test --color --include property
        env:
          STREAMDATA_RUNS: 500

      - name: Ping Vigilmon heartbeat
        if: success()
        run: curl -fsS "$VIGILMON_STREAMDATA_HEARTBEAT_URL"

Tag your property tests to run them separately from unit tests:

@tag :property
property "parse and serialize are inverse operations" do
  # ...
end

And configure ExUnit to include them in CI:

# test/test_helper.exs
ExUnit.configure(exclude: [:property])
ExUnit.start()

Step 6: Create a Heartbeat Monitor in Vigilmon

1. Sign in at vigilmon.online
2. Click New Monitor → Heartbeat
3. Name it StreamData Property Tests — main branch
4. Set the expected interval to 25 hours (matches the daily CI schedule)
5. Copy the URL and store it as VIGILMON_STREAMDATA_HEARTBEAT_URL in CI secrets

If property tests stop running — because CI was cancelled, a generator crashed, or a property found a counterexample — no heartbeat fires. Vigilmon alerts you within the missed-interval window.

Step 7: Reproduce Failures Deterministically

When a property test fails, StreamData prints the seed used to generate the counterexample. Capture and store it:

property "parse never raises" do
  check all input <- binary(),
            initial_seed: {1, 2, 3},   # fix seed to reproduce a specific failure
            max_runs: 1_000 do
    MyApp.Parser.parse(input)
  end
end

Add a CI step that logs the seed on failure:

- name: Run property tests (with seed in output)
  run: mix test --seed 0 --color --include property
  continue-on-error: true

Step 8: Alerting

In Vigilmon, go to Notifications → New Channel → Slack:

Heartbeat missed:

🔴 MISSED: StreamData Property Tests — main branch
Last successful property run: 30 hours ago
Action: Check CI — generator crash, property violation, or shrinking timeout

What You Built

| What | How | |------|-----| | Round-trip property | Verifies parse → serialize → parse is identity | | Crash-safety property | Ensures no binary input raises in parse/1 | | Custom generators | Typed generators for domain structs | | Generator smoke test | Mix task validates generators produce 50 samples | | Generator tests | ExUnit properties on the generators themselves | | CI integration | GitHub Actions daily schedule with max_runs: 500 | | CI heartbeat | Vigilmon heartbeat — alerts when property tests miss | | Slack alerting | Vigilmon Slack notification channel |

StreamData finds the bugs your example tests miss. Vigilmon ensures StreamData keeps running.

Next Steps

• Use StreamData.scale/2 to generate larger inputs for scalability properties without slowing the default test run
• Add @tag timeout: 120_000 to property tests with deep shrinking to prevent CI timeout kills
• Combine StreamData with ExMachina: generate random factory params to stress-test your Ecto changesets
• Set up a Vigilmon status page for your QA team showing property test health over time

Get started free at vigilmon.online.