← Interview Prep

Python — Automation Workflows

Build-style mock tasks that reproduce the shape of a real network-automation tool — gNMI telemetry, Ansible variable precedence, a Nornir-style deploy pipeline — with the network mocked out so the logic stands alone.

These are not pure-algorithm puzzles. Each one asks you to assemble a small pipeline that looks like a real tool: ingest structured data, transform it, compare, and act. There is no network — the streams, inventories, and devices are mocked so the interviewer can watch how you structure the flow, name the stages, and keep each function pure and testable.

The common thread: parse into a normalized shape, then compute over that shape. Flatten nested notifications before you rate them; resolve variables before you render; build an intended config before you diff. Keep the transform steps small and side-effect-free, and push the one impure action (applying a patch) to the very edge.

gNMI telemetry: flatten notifications & compute bit-rates

Problem statement. A pygnmi-style client hands you nested notifications; all of the pain is in parsing them and computing a rate from the counters. There is no network here: the mock MOCK_STREAM returns data in exactly the form of pygnmi's client.subscribe()/get(). Only the standard library is allowed; target time is 15 minutes.

A pygnmi-style notification looks like:

{"timestamp": <ns>, "prefix": "interfaces",
 "update": [{"path": "interface[name=eth0]/state/counters/in-octets",
             "val": 1000}, ...]}

1. flatten(notification) -> dict[str, int] — join prefix + path into one full path and return {full_path: val}. For example flatten(NOTIF0) yields {"interfaces/interface[name=eth0]/state/counters/in-octets": 1000, ...}.
2. bps(samples) -> dict[str, float] — samples is a list of (timestamp_ns, flat_dict) ordered by time. For every path present in both the first and the last sample, return the rate in bits per second: (last - first) * 8 / dt_seconds. Skip a path that appears only once. Empty input returns {}.

Follow-up. 64-bit counters do not wrap in practice, but old 32-bit SNMP counters do — how do you tell a wrap apart from a device restart? And why is a subscribe STREAM (on-change / sample) better than polling with a periodic get?

Approach

• Normalize first. flatten is the ingest stage: it turns one nested notification into a flat {path: value} map. Concatenate the prefix with each update's path (guarding the empty-prefix case) and read the val. Keep it pure — no rate math here.
• Rate over samples. bps only needs the first and last samples: a rate is a difference over an elapsed time, so the middle points do not matter for an average rate. Convert nanoseconds to seconds once.
• Guard the edges. Fewer than two samples, or a non-positive dt, means there is no rate to compute — return {}. Iterate the first sample's paths and only emit those also present in the last, which naturally drops paths seen once.
• Composition. In a real tool you would map flatten over the stream to build the samples list, then feed it to bps — two small stages, each independently testable.

Solution (Python)

"""Drill 13 (solution): flatten pygnmi notifications and compute bit-rates."""
from __future__ import annotations

MOCK_STREAM = [
    {
        "timestamp": 0,
        "prefix": "interfaces",
        "update": [
            {"path": "interface[name=eth0]/state/counters/in-octets", "val": 1000},
            {"path": "interface[name=eth0]/state/counters/out-octets", "val": 5000},
        ],
    },
    {
        "timestamp": 1_000_000_000,
        "prefix": "interfaces",
        "update": [
            {"path": "interface[name=eth0]/state/counters/in-octets", "val": 2000},
            {"path": "interface[name=eth0]/state/counters/out-octets", "val": 5500},
        ],
    },
]


def flatten(notification: dict) -> dict[str, int]:
    prefix = notification.get("prefix", "")
    out: dict[str, int] = {}
    for upd in notification.get("update", []):
        path = upd["path"]
        full = f"{prefix}/{path}" if prefix else path
        out[full] = upd["val"]
    return out


def bps(samples: list[tuple[int, dict]]) -> dict[str, float]:
    if len(samples) < 2:
        return {}
    ts_first, first = samples[0]
    ts_last, last = samples[-1]
    dt = (ts_last - ts_first) / 1e9
    if dt <= 0:
        return {}
    out: dict[str, float] = {}
    for path, start in first.items():
        if path in last:
            out[path] = (last[path] - start) * 8 / dt
    return out

Ansible-style rendering: variable precedence & {{ }} substitution

Problem statement. The classic source of Ansible bugs is variable precedence. There is no Jinja and no network here: the inventory is a plain dict (the mock INVENTORY). Standard library only; target time is 12 minutes.

The inventory shape:

{"defaults": {...},
 "groups": {"<g>": {"vars": {...}}, ...},
 "hosts":  {"<h>": {"vars": {...}, "groups": ["g1","g2"]}, ...}}

1. resolve_vars(inventory, host) -> dict — merge variables in ascending priority: defaults < group_vars (in the host's group order) < host_vars. A later group overrides an earlier one; host_vars override everything. An unknown host raises KeyError.
2. render(template, variables) -> str — replace placeholders of the form {{ name }} (inner whitespace does not matter) with values from variables. A missing variable raises KeyError(name).

Follow-up. Where do extra-vars (-e) sit in this chain in real Ansible? (At the very top — they override everything.) And why is group_vars/all dangerous?

Approach

• Layer the merge. resolve_vars is just dict.update applied in the right order: start from a copy of defaults, apply each group's vars in the host's listed group order, then apply host vars last. Because later updates win, this is the precedence ladder — no special-casing needed.
• Validate up front. Look the host up first and raise KeyError(host) if absent, before doing any merging. Copy defaults so you never mutate the inventory.
• Render with one regex. A single pattern \{\{\s*(\w+)\s*\}\} captures the name and tolerates surrounding whitespace. Use re.sub with a callback so a missing key raises KeyError at the exact placeholder, and coerce values with str() so ints render.
• Two composable steps. Resolve first, render second — the same order Ansible itself uses: build the fact/var namespace, then template against it.

Solution (Python)

"""Drill 14 (solution): ansible-like variable precedence and rendering."""
from __future__ import annotations

import re

_PLACEHOLDER = re.compile(r"\{\{\s*(\w+)\s*\}\}")

INVENTORY = {
    "defaults": {"ntp": "10.0.0.1", "mtu": 1500, "domain": "corp.net"},
    "groups": {
        "emea": {"vars": {"ntp": "10.1.0.1"}},
        "edge": {"vars": {"mtu": 9000}},
    },
    "hosts": {
        "r1": {"vars": {"hostname": "r1"}, "groups": ["emea", "edge"]},
        "r2": {"vars": {"hostname": "r2"}, "groups": ["emea"]},
    },
}


def resolve_vars(inventory: dict, host: str) -> dict:
    hosts = inventory["hosts"]
    if host not in hosts:
        raise KeyError(host)
    result = dict(inventory.get("defaults", {}))
    host_entry = hosts[host]
    for group in host_entry.get("groups", []):
        group_vars = inventory.get("groups", {}).get(group, {}).get("vars", {})
        result.update(group_vars)
    result.update(host_entry.get("vars", {}))
    return result


def render(template: str, variables: dict) -> str:
    def repl(match: re.Match) -> str:
        name = match.group(1)
        if name not in variables:
            raise KeyError(name)
        return str(variables[name])

    return _PLACEHOLDER.sub(repl, template)

Nornir-style deploy pipeline: SoT → generate → diff → patch → deploy

Problem statement. A full network-as-code cycle with no network: source-of-truth → generate the intended config from a template → compare against running → patch → deploy to a mock device. MockDevice plays the role of the connection (napalm/netmiko). Standard library only; target time is 15 minutes.

MockDevice exposes:

.get_config() -> str          # current running config (line by line)
.apply(patch: list[str])      # "no X" removes line X, otherwise adds it

Implement:

1. generate_config(sot, host, template) -> str — take the host's vars from sot[host] and substitute them into the {{ name }} template.
2. config_diff(running, intended) -> dict{"added":[...], "removed":[...]} — added is the lines in intended not in running (in intended order); removed is the lines in running not in intended (in running order). Ignore blank lines.
3. make_patch(diff) -> list[str] — first remove the excess: "no " + line for each removed line, then add each added line as-is.
4. deploy(device, sot, host, template) -> dict{"changed": bool, "patch": [...]} — build intended, pull running, compute diff, patch, apply. changed == (patch is non-empty). A repeated deploy returns changed=False, patch=[].

Follow-up. How does this differ from napalm ... commit with atomic rollback? Why is idempotency a mandatory property? And where is the dry-run in this design?

Approach

• Four small stages, one edge effect. Generate, diff, and patch are all pure functions of their inputs; only device.apply mutates state. Keeping the impure step at the very end is what makes the pipeline testable and gives you a natural dry-run (stop before apply).
• Reuse the renderer. generate_config is the same {{ name }} substitution as the Ansible drill — resolve the host's vars from the SoT and template against them.
• Diff on normalized lines. A shared _lines helper strips whitespace and drops blanks so both sides compare apples to apples. Build sets for O(1) membership, but emit added/ removed in the original list order for readable, deterministic patches.
• Order the patch: remove then add. Removing stale lines before adding new ones avoids conflicts and mirrors how CLI config sessions apply no commands first. Because the diff of an already-converged device is empty, the patch is empty and deploy reports changed=False — idempotency falls out of the design.

Solution (Python)

"""Drill 15 (solution): nornir-like SoT -> render -> diff -> patch -> deploy."""
from __future__ import annotations

import re

_PLACEHOLDER = re.compile(r"\{\{\s*(\w+)\s*\}\}")

SOT = {"r1": {"hostname": "r1", "mtu": 9000}}
TEMPLATE = "hostname {{ hostname }}\nmtu {{ mtu }}\nservice timestamps"


class MockDevice:
    def __init__(self, running: str = ""):
        self._lines = [ln.strip() for ln in running.split("\n") if ln.strip()]

    def get_config(self) -> str:
        return "\n".join(self._lines)

    def apply(self, patch: list[str]) -> None:
        for cmd in patch:
            if cmd.startswith("no "):
                target = cmd[3:]
                self._lines = [ln for ln in self._lines if ln != target]
            elif cmd not in self._lines:
                self._lines.append(cmd)


def generate_config(sot: dict, host: str, template: str) -> str:
    variables = sot[host]

    def repl(match: re.Match) -> str:
        name = match.group(1)
        if name not in variables:
            raise KeyError(name)
        return str(variables[name])

    return _PLACEHOLDER.sub(repl, template)


def _lines(text: str) -> list[str]:
    return [ln.strip() for ln in text.split("\n") if ln.strip()]


def config_diff(running: str, intended: str) -> dict:
    run = _lines(running)
    want = _lines(intended)
    run_set, want_set = set(run), set(want)
    added = [ln for ln in want if ln not in run_set]
    removed = [ln for ln in run if ln not in want_set]
    return {"added": added, "removed": removed}


def make_patch(diff: dict) -> list[str]:
    patch = [f"no {ln}" for ln in diff["removed"]]
    patch.extend(diff["added"])
    return patch


def deploy(device: MockDevice, sot: dict, host: str, template: str) -> dict:
    intended = generate_config(sot, host, template)
    running = device.get_config()
    patch = make_patch(config_diff(running, intended))
    device.apply(patch)
    return {"changed": bool(patch), "patch": patch}