Skip to content

Latest commit

 

History

History
289 lines (227 loc) · 9.35 KB

File metadata and controls

289 lines (227 loc) · 9.35 KB

Testing Guide

SnmpKit's simulator lets you test SNMP code without hardware: real UDP, real PDUs, real v1/v2c semantics, all on localhost. This guide shows the patterns the library's own suite uses.

A device per test

defmodule MyApp.PollerTest do
  use ExUnit.Case, async: true

  alias SnmpKit.SnmpSim.ProfileLoader

  setup do
    # Bundled profiles: :cable_modem, :router, :switch
    {:ok, profile} = ProfileLoader.load_profile(:router)
    port = free_udp_port()
    {:ok, device} = SnmpKit.Sim.start_device(profile, port: port)
    %{target: "127.0.0.1:#{port}", device: device}
  end

  test "reads sysDescr", %{target: target} do
    assert {:ok, %{value: "Cisco ISR 2901 Integrated Services Router", type: :octet_string}} =
             SnmpKit.SNMP.get(target, "sysDescr.0")
  end

  defp free_udp_port do
    {:ok, socket} = :gen_udp.open(0)
    {:ok, port} = :inet.port(socket)
    :gen_udp.close(socket)
    port
  end
end

start_device/2 links the device to the test process, so it goes away with the test and async: true is safe as long as each test picks its own port.

Defining what the device answers

Manual profiles

{:ok, profile} =
  ProfileLoader.load_profile(:test_switch, {:manual, %{
    "1.3.6.1.2.1.1.1.0" => "Test Switch v1.0",                   # STRING
    "1.3.6.1.2.1.1.3.0" => %{type: "TimeTicks", value: 12345},   # explicit type
    "1.3.6.1.2.1.2.1.0" => 2,                                    # INTEGER
    "1.3.6.1.2.1.2.2.1.2.1" => "FastEthernet0/1",
    "1.3.6.1.2.1.2.2.1.2.2" => "FastEthernet0/2",
    "1.3.6.1.2.1.2.2.1.8.1" => 1,
    "1.3.6.1.2.1.2.2.1.8.2" => 2
  }})

Strings become OCTET STRING, integers INTEGER; use %{type: "Counter32" | "Gauge32" | "TimeTicks" | "IpAddress" | ..., value: v} for anything else. SnmpKit.Sim.start_device(%{objects: map}, port: p) accepts the same map (OID keys as strings or lists) without a profile step. sysUpTime.0 is always the device's own uptime since it started, whatever value the profile carries for it.

Recorded walks

Record a real device once and replay it forever, with SnmpKit itself or with net-snmp:

{:ok, _count} = SnmpKit.Sim.record("192.168.1.1", "test/fixtures/my_switch.walk", community: "public")
snmpwalk -v2c -c public -On 192.168.1.1 .1 > test/fixtures/my_switch.walk
{:ok, profile} = ProfileLoader.load_profile(:my_switch, {:walk_file, "test/fixtures/my_switch.walk"})

Both -On numeric output and named output are understood.

Values over time

sysUpTime.0 is always the device's own uptime. Counters grow with that uptime when their profile entry carries a counter behavior: a {:walk_file, _} profile gets one per counter from SnmpKit.SnmpSim.MIB.BehaviorAnalyzer unless you pass behaviors: yourself, so ifInOctets and friends on the bundled walks advance between polls; a {:manual, _} profile is static unless you pass behaviors: (:realistic_counters, :daily_patterns, {:increment_counters, cfg}, ...; see SnmpKit.SnmpSim.BehaviorConfig). Two polls of a walk-file device a few seconds apart therefore give SnmpKit.SNMP.Rate.rates/3 a real delta. For an exact wrap or a fixed delta, build the samples by hand:

t0 = %{type: :counter32, value: 4_294_967_000}
t1 = %{type: :counter32, value: 1_000}
{:ok, 129.6} = SnmpKit.SNMP.Rate.rate(t0, t1, 10_000)

Error injection (SnmpKit.SnmpSim.ErrorInjector) draws from the device process's random state; start the device with seed: to replay the same timeout/loss sequence run after run.

SNMPv3

A device started with v3_users: answers SNMPv3 with the User-based Security Model: engine discovery, authentication, privacy, time windows and the usmStats reports a manager expects.

{:ok, device} =
  SnmpKit.Sim.start_device(profile,
    port: port,
    v3_users: [
      %{name: "guest"},
      %{name: "ops", auth: :sha256, auth_password: "auth-secret"},
      %{name: "admin", auth: :sha256, auth_password: "auth-secret", priv: :aes128, priv_password: "priv-secret"}
    ]
  )

{:ok, _} = SnmpKit.SNMP.get(target, "sysDescr.0",
  version: :v3, security_name: "admin",
  auth_protocol: :sha256, auth_password: "auth-secret",
  priv_protocol: :aes128, priv_password: "priv-secret")

The engine id derives from the device id (pass engine_id: to fix it); a user must be addressed at least at its own security level.

Error conditions

SnmpKit.SnmpSim.ErrorInjector attaches to a running device:

{:ok, device} = SnmpKit.Sim.start_device(profile, port: port)
{:ok, injector} = SnmpKit.SnmpSim.ErrorInjector.start_link(device, port)

SnmpKit.SnmpSim.ErrorInjector.inject_timeout(injector, probability: 1.0, duration_ms: 5_000)
SnmpKit.SnmpSim.ErrorInjector.inject_packet_loss(injector, loss_rate: 0.3)
SnmpKit.SnmpSim.ErrorInjector.inject_snmp_error(injector, :genErr, target_oids: ["1.3.6.1.2.1.1.5.0"])
SnmpKit.SnmpSim.ErrorInjector.simulate_device_failure(injector, :reboot, duration_ms: 2_000)
SnmpKit.SnmpSim.ErrorInjector.clear_all_errors(injector)

Some conditions need no injector at all:

Condition How
unreachable device query a port nothing listens on: {:error, :timeout}
unknown object ask for an OID the profile lacks: {:error, :no_such_object} (v1: :no_such_name)
read-only agent SnmpKit.SNMP.set/4 against a simulated device: {:error, :not_writable}
wrong community pass another community:; the device drops the packet, so {:error, :timeout}

Many devices

{:ok, devices} =
  SnmpKit.Sim.start_device_population(
    [{:cable_modem, {:walk_file, "priv/walks/cable_modem.walk"}, count: 50},
     {:switch, {:walk_file, "priv/walks/switch.walk"}, count: 5}],
    port_range: 30_000..30_099
  )

targets = Enum.map(devices, &{&1.target, "sysUpTime.0"})
results = SnmpKit.SNMP.get_multi(targets, max_concurrent: 50)
assert Enum.all?(results, &match?({:ok, _}, &1))

Population devices run under the simulator's supervisor; stop them with SnmpKit.SnmpSim.stop_device/1 or SnmpKit.SnmpSim.stop/0. For whole environments, SnmpKit.SnmpSim.start/1 takes a configuration map or file (SnmpKit.SnmpSim.sample_config/0 shows the shape).

Testing code that calls SnmpKit without the network

Hide SnmpKit behind a small module in your application and swap it in tests:

defmodule MyApp.SNMPClient do
  @callback get(String.t(), String.t(), keyword()) :: {:ok, map()} | {:error, term()}
  def get(target, oid, opts \\ []), do: impl().get(target, oid, opts)
  defp impl, do: Application.get_env(:my_app, :snmp_client, SnmpKit.SNMP)
end

# test_helper.exs
Mox.defmock(MyApp.SNMPClientMock, for: MyApp.SNMPClient)
Application.put_env(:my_app, :snmp_client, MyApp.SNMPClientMock)
test "reports unreachable devices" do
  expect(MyApp.SNMPClientMock, :get, fn _target, _oid, _opts -> {:error, :timeout} end)
  assert {:error, :device_unreachable} = MyApp.DeviceMonitor.status("10.0.0.1")
end

Prefer the simulator for anything that depends on real SNMP behaviour (walk termination, table shapes, v1 versus v2c errors); use a mock only for pure control-flow tests.

Real devices

defmodule MyApp.RealDeviceTest do
  use ExUnit.Case
  @moduletag :real_device

  @target System.get_env("SNMP_TEST_TARGET")

  setup_all do
    if @target, do: :ok, else: {:skip, "set SNMP_TEST_TARGET to run"}
  end

  test "answers sysDescr" do
    assert {:ok, %{value: descr}} = SnmpKit.SNMP.get(@target, "sysDescr.0", timeout: 5_000)
    assert is_binary(descr)
  end
end

Exclude the tag by default in test_helper.exs and include it on demand with mix test --include real_device.

Performance tests

@moduletag :performance

test "walks ifTable quickly", %{target: target} do
  {micros, {:ok, rows}} = :timer.tc(fn -> SnmpKit.SNMP.walk(target, "ifTable") end)
  assert rows != []
  assert micros < 1_000_000
end

Keep wall-clock assertions behind an opt-in tag; they flake under a loaded full run. SnmpKit.SNMP.benchmark_device/3 measures how a device responds to different max_repetitions values if you want data rather than a pass/fail.

Test configuration

# config/test.exs
config :snmpkit,
  timeout: 1_000,   # fail fast against simulated devices
  retries: 0

config :logger, level: :warning

Tags used by SnmpKit's own suite

Tag Purpose
:unit, :integration, :snmpv3, :protocol, :slow, :docsis run by default; select one with mix test --only <tag>
:performance timing assertions
:memory memory-intensive MIB tests
:mib_oracle MIB parser cross-check against libsmi / net-snmp
:needs_simulator simulator tests bound to fixed ports (would clash in a parallel run)
:manual, :real_device need hardware or a person

Everything from :performance down is excluded in test/test_helper.exs; include a tag with mix test --include <tag>.

Continuous integration

# .github/workflows/test.yml
jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        elixir: ["1.18.4", "1.20.0"]
        otp: ["28.0"]
    steps:
      - uses: actions/checkout@v4
      - uses: erlef/setup-beam@v1
        with:
          elixir-version: ${{ matrix.elixir }}
          otp-version: ${{ matrix.otp }}
      - run: mix deps.get
      - run: mix compile --warnings-as-errors
      - run: mix test

Simulated devices bind localhost UDP ports, which every hosted runner allows.