NetAuto Academy

Nornir for Network Engineers

Capstone: A Nornir-Based Network Health Check

Advanced18 min read

What you'll learn

  • Combining inventory, filtering, task execution, and concurrency into one script
  • Where a real device (Option A, nornir_netmiko) fits versus the local, no-device path (Option B)
  • Reading a real, verified end-to-end run and its JSON report
  • What's next: NAPALM, for multi-vendor abstraction beyond Cisco IOS
🎓 Teach This Lesson

Listen to a guided explanation of this lesson.

Spoken narration isn't supported in this browser.

What this capstone combines

Every lesson in this module built one piece. This capstone runs all of them together, against the exact same VLAN desired-state data this curriculum has used since Phase 2.9.

  1. InitNornir()Lesson 3 — load hosts.yaml/groups.yaml/defaults.yaml
  2. Inject credentialsLesson 2 — from environment variables, never on disk
  3. nr.filter(site="HQ")Lesson 4 — demonstrated, not required for the full run
  4. nr.run(task=check_vlans)Lesson 3 + 6 — concurrent by default
  5. Option A: nornir_netmikoLesson 5 — real device
  6. Option B: simulate_connection()no device — this capstone's default path
  7. AggregatedResult -> JSON reportLesson 3 — structured results

Two options, same shape as every prior capstone

Option A: swap check_vlans's body for nr.run(task=netmiko_send_command, command_string="show vlan brief") against real devices, exactly as Lesson 5 verified. Option B (used below): simulate_connection(), the same Netmiko-shaped simulation pattern from Your First Network Automation -- no real device, but the exact same Nornir workflow around it.

Quick checkConcept check
What's the actual difference between Option A and Option B in this capstone?

The full script

nornir_health_check.py
import json
import os
import time
from nornir import InitNornir
from nornir.core.task import Task, Result

REQUIRED_VLANS = {
  "SW1": [{"id": 10, "name": "USERS"}, {"id": 20, "name": "VOICE"}],
  "SW2": [{"id": 10, "name": "USERS"}],
  "SW3": [{"id": 10, "name": "USERS"}, {"id": 30, "name": "MGMT"}],
  "SW4": [{"id": 10, "name": "USERS"}],
  "SW5": [{"id": 10, "name": "USERS"}],
}

_SCENARIOS = {
  "SW1": {"vlans": [{"id": 10, "name": "USERS"}, {"id": 20, "name": "VOICE"}], "outcome": "ok"},
  "SW2": {"vlans": [], "outcome": "ok"},
  "SW3": {"vlans": [{"id": 10, "name": "USERS"}], "outcome": "ok"},
  "SW4": {"vlans": [], "outcome": "unreachable"},
  "SW5": {"vlans": [], "outcome": "auth-failed"},
}

def simulate_connection(hostname):
  # Option B: the same Netmiko-shaped simulation pattern from
  # Your First Network Automation -- no real device, same workflow.
  time.sleep(0.3)
  scenario = _SCENARIOS[hostname]
  if scenario["outcome"] == "unreachable":
      raise ConnectionError(f"{hostname} did not respond")
  if scenario["outcome"] == "auth-failed":
      raise PermissionError(f"{hostname} rejected the credentials")
  return scenario["vlans"]

def check_vlans(task: Task) -> Result:
  hostname = task.host.name
  try:
      actual_vlans = simulate_connection(hostname)
  except (ConnectionError, PermissionError) as e:
      return Result(host=task.host, result=str(e), failed=True)

  required = REQUIRED_VLANS[hostname]
  missing = [v for v in required if v not in actual_vlans]
  if missing:
      return Result(
          host=task.host,
          result=f"missing VLANs: {[v['id'] for v in missing]}",
          failed=False,
          changed=True,
      )
  return Result(host=task.host, result="already compliant", failed=False, changed=False)

nr = InitNornir(
  inventory={
      "plugin": "SimpleInventory",
      "options": {
          "host_file": "inventory/hosts.yaml",
          "group_file": "inventory/groups.yaml",
          "defaults_file": "inventory/defaults.yaml",
      },
  },
)
nr.inventory.defaults.username = os.environ.get("NORNIR_USERNAME", "")
nr.inventory.defaults.password = os.environ.get("NORNIR_PASSWORD", "")

hq_only = nr.filter(site="HQ")
print(f"HQ-only run would target: {list(hq_only.inventory.hosts.keys())}")

agg = nr.run(task=check_vlans)

report = {"compliant": [], "non_compliant": [], "unreachable": [], "auth_failed": []}
for hostname, multi_result in agg.items():
  r = multi_result[0]
  if not r.failed and not r.changed:
      report["compliant"].append({"host": hostname, "detail": r.result})
  elif not r.failed and r.changed:
      report["non_compliant"].append({"host": hostname, "detail": r.result})
  elif "did not respond" in r.result:
      report["unreachable"].append({"host": hostname, "detail": r.result})
  else:
      report["auth_failed"].append({"host": hostname, "detail": r.result})

print("--- Summary ---")
print(f"Checked: {len(agg)} devices")
print(f"compliant: {len(report['compliant'])}")
print(f"non-compliant: {len(report['non_compliant'])}")
print(f"unreachable: {len(report['unreachable'])}")
print(f"auth-failed: {len(report['auth_failed'])}")

with open("nornir_report.json", "w") as f:
  json.dump(report, f, indent=2)
print("Report written to nornir_report.json")

The complete, real, verified capstone script. Reuses the exact REQUIRED_VLANS desired-state data from Phase 2.9 onward, and the simulate_connection() philosophy from Your First Network Automation.

Quick checkIdentify the correct code
Why does check_vlans check both `failed` and `changed` on the returned Result, rather than just `failed`?

Real, verified end-to-end output

terminal output — real, verified run
HQ-only run would target: ['SW1', 'SW2', 'SW3']
--- Summary ---
Checked: 5 devices
compliant: 1
non-compliant: 2
unreachable: 1
auth-failed: 1
Report written to nornir_report.json

Real, verified output from an actual run of the script above. The HQ-only filter line demonstrates Lesson 4's capability without narrowing the actual run -- the full check still covers all 5 devices.

nornir_report.json — real, verified content
{
"compliant": [
  { "host": "SW1", "detail": "already compliant" }
],
"non_compliant": [
  { "host": "SW2", "detail": "missing VLANs: [10]" },
  { "host": "SW3", "detail": "missing VLANs: [30]" }
],
"unreachable": [
  { "host": "SW4", "detail": "SW4 did not respond" }
],
"auth_failed": [
  { "host": "SW5", "detail": "SW5 rejected the credentials" }
]
}

Real, verified file content -- grep-confirmed to contain no credential of any kind, matching the runtime-injection pattern from Lesson 2.

Quick checkPredict output
Based on the verified output above, why does SW3 land in non_compliant rather than compliant?
Quick checkScenario
A learner wants to run this exact script against real switches instead of the simulation. What's the minimal, correct change, per this module's Option A pattern?

What's next

One vendor's CLI so far

Every device in this curriculum has spoken Cisco IOS. NAPALM, the next planned module, addresses a different problem: a single, consistent API across multiple vendors (Cisco, Juniper, Arista, and others) instead of vendor-specific commands. Nornir itself doesn't change -- nornir_napalm is simply a different task plugin, the same role nornir_netmiko has played throughout this module.

Quick checkConcept check
What problem does NAPALM (the next module) address that this module doesn't?

Key takeaways

  • This capstone combines every piece of the module: InitNornir(), runtime credential injection, nr.filter(), nr.run() with concurrency, and AggregatedResult-based reporting -- all verified together in one real, runnable script.
  • Option A (nornir_netmiko against a real device) and Option B (simulate_connection(), used here) differ only in check_vlans's body -- the surrounding Nornir workflow is identical.
  • Real, verified run: 1 compliant, 2 non-compliant, 1 unreachable, 1 auth-failed across the 5-device fleet -- written to nornir_report.json with zero credentials on disk.
  • Next: NAPALM, for a consistent API across multiple vendors -- a different task plugin, the same Nornir workflow.