Five whiteboard-style Python exercises drawn from a network-engineering / infra live-coding round: config parsing, prefix math, structure flattening, idempotent diffs, and asyncio orchestration. Each has a problem statement, an approach with complexity, a full reference solution, and what the interviewer is really testing.
These are short, self-contained tasks (roughly 15–25 minutes each) that lean on the standard library. The point is rarely a clever algorithm — it is clean parsing, correct edge-case handling, determinism, and being able to explain trade-offs out loud. Read the statement, try it yourself, then compare against the reference.
Time budget: ~15 min. Implement parse_interfaces(config) -> dict[str, dict].
Input: configuration text in Arista/Cisco IOS style.
Output: a dict {interface_name: {...}} in order of appearance in the config.
For each interface produce:
description : str | None — everything after description mtu : int | Noneipv4 : str | None — exactly as written in the config, e.g. "10.0.0.1/31"enabled : bool — False if a shutdown line is present, defaults to Truevlans : list[int] — the expanded list from
switchport trunk allowed vlan 10,20,30-32 → [10, 20, 30, 31, 32]; defaults to []Rules:
! lines, blank lines, and #... comments are ignored;interface X block are ignored;Single linear pass over the lines with a small state machine: track the current interface block,
seed each new block with default values, and match subordinate lines with a handful of precompiled regexes.
VLAN specs are expanded by splitting on commas and unrolling a-b ranges. Keeping insertion order
is free — a plain dict preserves it. Complexity is O(n) in the number of lines
(VLAN expansion is O(total vlans)); memory is O(size of the parsed result).
"""Task 1 (reference solution): parse a CLI config into structured data."""
from __future__ import annotations
import re
from typing import Any
IFACE_RE = re.compile(r"^interface\s+(\S+)\s*$")
DESC_RE = re.compile(r"^description\s+(.+?)\s*$")
MTU_RE = re.compile(r"^mtu\s+(\d+)\s*$")
IPV4_RE = re.compile(r"^ip address\s+(\S+)\s*$")
VLAN_RE = re.compile(r"^switchport (?:trunk allowed vlan|access vlan)\s+(\S+)\s*$")
def _expand_vlans(spec: str) -> list[int]:
out: list[int] = []
for chunk in spec.split(","):
chunk = chunk.strip()
if not chunk:
continue
if "-" in chunk:
start, end = chunk.split("-", 1)
out.extend(range(int(start), int(end) + 1))
else:
out.append(int(chunk))
return sorted(set(out))
def parse_interfaces(config: str) -> dict[str, dict[str, Any]]:
result: dict[str, dict[str, Any]] = {}
current: dict[str, Any] | None = None
for raw in config.splitlines():
line = raw.strip()
if not line or line == "!" or line.startswith("#"):
continue
m = IFACE_RE.match(line)
if m:
current = {
"description": None,
"mtu": None,
"ipv4": None,
"enabled": True,
"vlans": [],
}
result[m.group(1)] = current
continue
if current is None:
continue
if m := DESC_RE.match(line):
current["description"] = m.group(1)
elif m := MTU_RE.match(line):
current["mtu"] = int(m.group(1))
elif m := IPV4_RE.match(line):
current["ipv4"] = m.group(1)
elif line == "shutdown":
current["enabled"] = False
elif line == "no shutdown":
current["enabled"] = True
elif m := VLAN_RE.match(line):
current["vlans"] = _expand_vlans(m.group(1))
return result
interface block (current is None) must be dropped, not attached to a phantom interface.dict insertion order rather than sorting.a-b ranges, stray spaces, and de-duplicate; the reference also sorts for determinism.^...$ and \s*$ avoids matching substrings and trailing whitespace surprises.Time budget: ~15 min. The ipaddress module is allowed. Three functions:
summarize(prefixes: list[str]) -> list[str]["10.0.0.0/25", "10.0.0.128/25"] → ["10.0.0.0/24"].
IPv4 and IPv6 are handled separately: all IPv4 first (ascending), then IPv6.
An invalid prefix → ValueError.find_overlaps(prefixes: list[str]) -> list[tuple[str, str]](outer, inner), returned as a sorted list of tuples.
Different address families never overlap.longest_prefix_match(routes: list[str], addr: str) -> str | NoneNone.Interviewer follow-up: how would you make LPM O(32) instead of O(n)? (Answer: a binary trie / Patricia tree — be ready to describe the structure.)
Lean on ipaddress. Parse once with a helper that re-raises a clean ValueError.
For summarize, split by family and call collapse_addresses per family, which merges
adjacent/contained prefixes into the minimal set. For find_overlaps, do the naive
O(n²) pairwise scan using Network.overlaps, orienting each pair so the shorter
prefix length (the larger, containing network) is the "outer". For longest_prefix_match, scan all
routes and keep the containing network with the largest prefixlen — O(n) per lookup.
The follow-up trie brings lookups to O(address-bits) after O(n·bits) build.
"""Task 2 (reference solution): IP/prefix utilities."""
from __future__ import annotations
import ipaddress
from ipaddress import IPv4Network, IPv6Network
def _parse(prefixes: list[str]):
nets = []
for p in prefixes:
try:
nets.append(ipaddress.ip_network(p, strict=False))
except ValueError as exc:
raise ValueError(f"invalid prefix: {p}") from exc
return nets
def summarize(prefixes: list[str]) -> list[str]:
nets = _parse(prefixes)
v4 = [n for n in nets if isinstance(n, IPv4Network)]
v6 = [n for n in nets if isinstance(n, IPv6Network)]
out = list(ipaddress.collapse_addresses(v4)) + list(ipaddress.collapse_addresses(v6))
return [str(n) for n in out]
def find_overlaps(prefixes: list[str]) -> list[tuple[str, str]]:
nets = _parse(prefixes)
pairs: list[tuple[str, str]] = []
for i, a in enumerate(nets):
for b in nets[i + 1 :]:
if a.version != b.version:
continue
if a.overlaps(b):
if a.prefixlen <= b.prefixlen:
outer, inner = a, b
else:
outer, inner = b, a
pairs.append((str(outer), str(inner)))
return sorted(pairs)
def longest_prefix_match(routes: list[str], addr: str) -> str | None:
ip = ipaddress.ip_address(addr)
best = None
for net in _parse(routes):
if net.version != ip.version:
continue
if ip in net:
if best is None or net.prefixlen > best.prefixlen:
best = net
return str(best) if best else None
collapse_addresses raises on mixed input, and overlaps across families are meaningless.ValueError with useful context, using raise ... from exc.strict=False tolerates host bits set (e.g. 10.0.0.1/24) instead of rejecting them.Time budget: ~20 min. Two functions:
1) flatten(data, prefix="") -> dict[str, scalar] — recursively turn a nested
dict/list into flat gNMI paths.
{"interfaces": {"interface": [{"name": "Ethernet1",
"config": {"mtu": 9214}}]}}
->
{"/interfaces/interface[name=Ethernet1]/name": "Ethernet1",
"/interfaces/interface[name=Ethernet1]/config/mtu": 9214}
Rules:
dict → path + "/" + keyname/index/id exists
(checked in exactly that order) → path + "[name=<value>]"path + "[<index>]"prefix is given, it is used as the start of the path2) diff_flat(actual, intended) -> dict with keys added / removed / changed:
added : [(path, intended_value)] — present only in intendedremoved : [(path, actual_value)] — present only in actualchanged : [(path, actual_value, intended_value)]All three lists are sorted.
Follow-up: how would you turn this diff into a gNMI SetRequest (update / replace / delete)?
Classic recursive descent. For a dict, recurse into each key extending the path with /key.
For a list, prefer a stable key (name, then index, then id) so paths are
keyed by identity rather than positional index — falling back to [i] only when no scalar key
exists. Scalars terminate the recursion by writing into the output dict. The diff is then pure set arithmetic
over the two flat dictionaries: keys only in intended are added, keys only in actual are
removed, shared keys with differing values are changed. Flatten is O(nodes);
diff is O(paths) plus the sort.
"""Task 3 (reference solution): flatten YANG/gNMI-like structures and diff them."""
from __future__ import annotations
from typing import Any
KEY_CANDIDATES = ("name", "index", "id")
def _key_of(item: dict[str, Any]) -> tuple[str, Any] | None:
for k in KEY_CANDIDATES:
if k in item and not isinstance(item[k], (dict, list)):
return k, item[k]
return None
def flatten(data: Any, prefix: str = "") -> dict[str, Any]:
out: dict[str, Any] = {}
if isinstance(data, dict):
for key, value in data.items():
out.update(flatten(value, f"{prefix}/{key}"))
elif isinstance(data, list):
for idx, item in enumerate(data):
if isinstance(item, dict):
kv = _key_of(item)
if kv is not None:
out.update(flatten(item, f"{prefix}[{kv[0]}={kv[1]}]"))
continue
out.update(flatten(item, f"{prefix}[{idx}]"))
else:
out[prefix] = data
return out
def diff_flat(actual: dict[str, Any], intended: dict[str, Any]) -> dict[str, list]:
added = [(p, intended[p]) for p in intended if p not in actual]
removed = [(p, actual[p]) for p in actual if p not in intended]
changed = [
(p, actual[p], intended[p])
for p in intended
if p in actual and actual[p] != intended[p]
]
return {
"added": sorted(added),
"removed": sorted(removed),
"changed": sorted(changed),
}
name → index → id in that exact order, and only when the candidate value is a scalar (a nested dict/list named "id" is not a list key).[i].prefix argument rather than mutating shared state; scalars are the only base case that writes.update, removed → delete, and know when a subtree warrants replace vs per-leaf update.Time budget: ~15 min.
config_diff(actual: list[str], intended: list[str]) -> list[str] —
returns the commands that bring actual to intended:
no <line> for lines in actual that are absent from intended (in actual order);Normalization before comparison:
!, and lines starting with #.is_compliant(actual, intended) -> bool — True when the diff is empty.
The key point the interviewer wants to hear: why idempotency matters, and why "just push the whole config" is a bad strategy in production.
Normalize both sides into comparable canonical lines (strip, collapse whitespace, drop noise), then use
sets for membership tests while iterating the original ordered lists to preserve the required output
order. A seen set suppresses duplicates. Emit missing intended lines first, then negate stray
actual lines with a no prefix. is_compliant is just "diff is empty".
Complexity is O(n) with set lookups over the total number of lines.
"""Task 4 (reference solution): idempotent config diff for a flat CLI config."""
from __future__ import annotations
def _normalize(lines: list[str]) -> list[str]:
out = []
for line in lines:
line = line.strip()
if not line or line == "!" or line.startswith("#"):
continue
out.append(" ".join(line.split()))
return out
def config_diff(actual: list[str], intended: list[str]) -> list[str]:
act = _normalize(actual)
int_ = _normalize(intended)
act_set = set(act)
int_set = set(int_)
commands: list[str] = []
seen: set[str] = set()
for line in int_:
if line not in act_set and line not in seen:
commands.append(line)
seen.add(line)
for line in act:
if line not in int_set and line not in seen:
commands.append(f"no {line}")
seen.add(line)
return commands
def is_compliant(actual: list[str], intended: list[str]) -> bool:
return not config_diff(actual, intended)
seen set prevents emitting the same command twice.no <line> negation is a simplification — real gear has hierarchy and order-sensitive commands.Time budget: ~25 min. Standard library only.
async collect(devices, fetch, *, concurrency=5, retries=2,
timeout=5.0, backoff=0.0) -> dict[str, Result]
devices : list of device namesfetch : async callable (device) -> value (may raise or hang)Requirements:
concurrency fetch calls running at once (a Semaphore);timeout (asyncio.wait_for);retries + 1 attempts total; between attempts, pause
backoff * 2**attempt_number (with backoff=0, no pause);{device: Result}.Result (dataclass): device: str, ok: bool,
value: Any = None (on success), error: str | None
("timeout" on timeout, otherwise "TypeError: text"),
attempts: int (how many attempts were actually made).
Follow-up: how does this differ from a threading approach (Nornir/Netmiko)? When does asyncio not help (CPU-bound parsing, blocking drivers)?
Spawn one task per device with asyncio.create_task and gather them, so devices proceed
independently and one failure cannot sink the run. A shared Semaphore(concurrency) bounds the
number of in-flight fetch calls. Each per-device coroutine loops up to retries + 1
times, wrapping fetch in asyncio.wait_for(..., timeout), translating a
TimeoutError into "timeout" and any other exception into
"Type: message", and sleeping with exponential backoff between attempts. It returns a
Result either way. Concurrency is bounded by the semaphore; total wall-clock is roughly
ceil(len(devices)/concurrency) waves × per-call cost.
"""Task 5 (reference solution): concurrent collection with limits, retries, timeouts."""
from __future__ import annotations
import asyncio
from dataclasses import dataclass
from typing import Any, Awaitable, Callable
@dataclass
class Result:
device: str
ok: bool
value: Any = None
error: str | None = None
attempts: int = 0
async def _one(
device: str,
fetch: Callable[[str], Awaitable[Any]],
sem: asyncio.Semaphore,
retries: int,
timeout: float,
backoff: float,
) -> Result:
attempts = 0
last_error = "unknown error"
for attempt in range(retries + 1):
attempts += 1
try:
async with sem:
value = await asyncio.wait_for(fetch(device), timeout=timeout)
return Result(device=device, ok=True, value=value, attempts=attempts)
except asyncio.TimeoutError:
last_error = "timeout"
except Exception as exc: # noqa: BLE001 - one bad device must not kill the run
last_error = f"{type(exc).__name__}: {exc}"
if attempt < retries and backoff:
await asyncio.sleep(backoff * (2**attempt))
return Result(device=device, ok=False, error=last_error, attempts=attempts)
async def collect(
devices: list[str],
fetch: Callable[[str], Awaitable[Any]],
*,
concurrency: int = 5,
retries: int = 2,
timeout: float = 5.0,
backoff: float = 0.0,
) -> dict[str, Result]:
sem = asyncio.Semaphore(concurrency)
tasks = [
asyncio.create_task(_one(d, fetch, sem, retries, timeout, backoff))
for d in devices
]
results = await asyncio.gather(*tasks)
return {r.device: r for r in results}
fetch so a task holding the slot while sleeping on backoff doesn't starve others — and so the limit reflects actual in-flight calls.wait_for; a hung fetch must not block the whole run, and TimeoutError maps to the literal "timeout".Exception per device (never let one raise out of gather); shape the error string as "Type: message".attempts counts real tries; backoff is backoff * 2**attempt and is skipped after the final attempt and when backoff==0.{device: Result} map.