Setup, Schema & the Mock Cluster
Chapter 8 gave you the design. Now you build it, step by step, exactly like the earlier chapters' labs. This first lab lays the foundation: the project skeleton, the risk-classification schema that drives all safety, and a mock cluster so everything runs locally with zero cloud cost or risk. By the end you'll have passing tests and no API calls yet.
Learning objectives
- Create the capstone project structure alongside the course starter kit.
- Define
RiskClassandDiagnosis— the shapes everything else depends on. - Build a mock Kubernetes cluster from JSON fixtures (deterministic, offline).
- Write your first tests and run them green — before spending a cent on the API.
llm-course-starter/devops-agent/. You can build along by typing each file yourself (recommended — it's how you learn), or read the finished version there and run it. The labs explain why each piece is shaped the way it is.What you'll build across the four build labs intermediate
This is the map for the whole capstone. It shows the DevOps agent as a stack of four layers, one per lab (8a, 8b, 8c, 8d). You are building the bottom layer now; each later lab sits on top of the ones below it.
- Read it bottom-up. The top row (
8a) is what you build in this lesson: the risk schema, the mock cluster, and offline tests. It sits at the base of everything — nothing above works until this is solid. 8badds the tool registry + agent loop so the thing can actually look at the cluster (read-only) and reason about it — that turns the foundation into a diagnostician.8cadds RAG over runbooks (so answers are grounded in the company's real docs) and the policy gate (so risky actions are blocked or need approval), plus an audit trail.8dadds evals — automated tests of accuracy and safety that must pass — and only then swaps the mock for real infrastructure.- The blue line at the bottom is the golden rule: mock-first. Everything runs and is tested against a fake cluster before it is ever pointed at real infra.
In short: Each layer is proven green before the next is added. So when something breaks in 8c, you already trust 8a and 8b — the bug is almost certainly in the layer you just added.
| Lab | You build | Result |
|---|---|---|
| 8a (here) | Skeleton, risk schema, mock cluster, first tests | Foundation runs offline, tests green |
| 8b | Tool registry + the agent loop | A working read-only diagnostician |
| 8c | RAG over runbooks + the policy gate + audit | Grounded, safe, gated actions |
| 8d | Evals (incl. hard-fail safety) + go-real | Trustworthy, CI-gated, real-infra ready |
Step 1 · Create the project skeleton intermediate
Work inside the course starter kit so you reuse its virtualenv and .env.
terminalcd llm-course-starter
source .venv/bin/activate # same venv as chapters 1-6
pip install pytest # for the test cases (also in requirements.txt)
mkdir -p devops-agent/{agent,mock,runbooks,tests,evals}
cd devops-agent
touch agent/__init__.py mock/__init__.py # make them importable packages
Structure you're creating:
devops-agent/
├── agent/ # the brain: schema, tools, policy, rag, audit, loop
├── mock/ # fake cluster + actions (no real infra)
├── runbooks/# the company's docs (RAG source)
├── tests/ # safety tests — no API key needed
└── evals/ # accuracy + safety evals
Before writing any logic you lay down the folder structure. These terminal commands create the project inside the course starter kit so it reuses the same virtualenv and .env file you set up in earlier chapters — no new setup.
cd llm-course-starterthensource .venv/bin/activatemoves into the starter kit and turns on its virtual environment — an isolated Python with the course's libraries already installed.pip install pytestadds the test runner you'll use in Step 5.mkdir -p devops-agent/{agent,mock,runbooks,tests,evals}creates the project folder and its five sub-folders in one line. The{...}is shell shorthand that expands to five separatemkdirs.touch agent/__init__.py mock/__init__.pycreates empty__init__.pyfiles. That one detail is what makesagentandmockreal Python packages you canimportfrom — without them,from mock import clusterlater would fail.
What the output means: No visible output — success is silent. Afterwards the tree shown below exists: agent/ (the brain), mock/ (the fake cluster), runbooks/, tests/, and evals/.
Try this: Run ls devops-agent to confirm the five folders exist. If a later step says No module named 'mock', the missing __init__.py is the usual culprit — check it's there.
Step 2 · The risk schema — the spine of all safety intermediate
Before any tool or agent, define how risky an action is. Every tool will be tagged with one of these, and the safety gate (Lab 8c) reads the tag to decide what's allowed. This is the most important type in the whole project.
agent/schemas.pyfrom enum import Enum
from typing import Optional
from pydantic import BaseModel, Field
class RiskClass(str, Enum):
READ_ONLY = "read_only" # 🟢 inspect only
REVERSIBLE = "reversible" # 🟡 scale, restart, open PR
SIGNIFICANT = "significant" # 🟠 apply non-prod, merge PR
IRREVERSIBLE = "irreversible" # 🔴 delete, prod apply
class Diagnosis(BaseModel):
likely_cause: str
evidence: list[str] # the facts it's citing
suggested_fix: str
fix_risk: RiskClass # how risky the fix is
confidence: float = Field(ge=0.0, le=1.0)
runbook_ref: Optional[str] = None
(The finished file also has a ToolCallRecord for the audit log — see Lab 8c.)
This file defines the two data shapes the whole project is built around, using Pydantic — a library that lets you declare exactly what fields a piece of data has and what type each one is, then checks the data matches. Think of it as a typed form the rest of the code must fill in correctly.
class RiskClass(str, Enum)is an enum: a fixed set of allowed values. An action is exactly one ofREAD_ONLY,REVERSIBLE,SIGNIFICANT, orIRREVERSIBLE— the four risk levels, from 'just looking' (🟢) to 'delete / apply in prod' (🔴).class Diagnosis(BaseModel)is the structured answer the agent must produce.BaseModelmeans Pydantic will validate every field.- The fields spell out a complete diagnosis:
likely_cause(a sentence),evidence(alist[str]— the facts it cites),suggested_fix, andfix_riskwhich is aRiskClass— so the fix is tagged with its danger level. confidence: float = Field(ge=0.0, le=1.0)forces the number to sit between 0 and 1 (ge= greater-or-equal,le= less-or-equal).runbook_ref: Optional[str] = Noneis an optional pointer to a doc, defaulting to nothing.
What the output means: Nothing runs — this is a definition. But now anywhere in the code, a Diagnosis is guaranteed to have these fields with these types, or Pydantic raises an error before the bad data can spread.
Try this: Later the safety gate will read fix_risk to decide what's allowed. Because it's an enum, a typo like "irreversable" is impossible — that is the whole point of using an enum instead of a plain string.
if risk == RiskClass.IRREVERSIBLE. A typo-proof enum means an action can never accidentally fall through the safety check because someone wrote "irreversable".Step 3 · The mock cluster advanced
We build against a fake cluster first. This is not a shortcut — it's the professional way to develop something dangerous: deterministic, free, and incapable of breaking anything. The fixtures include a crash-looping pod (with the real error in its logs) and an OOMKilled pod, so the agent has real problems to solve.
- The fixture data — the fake cluster state.
mock/fixtures.json (excerpt)
{ "pods": [{ "name": "checkout-api-7d9f", "namespace": "staging", "status": "CrashLoopBackOff", "restarts": 7, "logs": [ "INFO connecting to postgres...", "ERROR password authentication failed for user 'checkout'", "ERROR startup probe failed, exiting" ] }], "deploys": [{"name": "checkout-api", "revision": 12, "changed": "updated DATABASE_URL secret ref", "when": "18 min ago"}] }Notice the story hidden in the data: the DB auth fails, and 18 minutes ago someone changed the DATABASE_URL secret. A good agent should connect those.
- The read-only accessor — this is what tools will call.
Illustrative fragment — defines demo values / files are needed before this runs standalone.
mock/cluster.py
import json, os _STATE = json.load(open(os.path.join(os.path.dirname(__file__), "fixtures.json"))) def get_pods(namespace="staging"): return [p for p in _STATE["pods"] if p["namespace"] == namespace] def get_logs(pod, namespace="staging", lines=20): p = get_pod(pod, namespace) if not p: return f"Error: pod '{pod}' not found" return "\n".join(p.get("logs", [])[-lines:]) # ... get_pod, get_events, recent_deploys (see the finished file)
The mock cluster is a fake Kubernetes cluster that lives entirely in a JSON file. It lets you develop something dangerous safely: it's deterministic (same answer every time), free, and physically cannot break anything real. Two files make it up — the data (fixtures.json) and the reader (cluster.py).
- fixtures.json is just the cluster's state written down: a list of
podsanddeploys. One pod is stuck inCrashLoopBackOffwith real error logs, and a deploy 18 minutes ago changed the DATABASE_URL secret — that's a deliberate story the agent should piece together. - In cluster.py,
_STATE = json.load(open(...))reads that JSON once into memory. Theos.path.dirname(__file__)part means it finds the file next to the module, no matter which folder you run from. get_pods(namespace="staging")returns the pods in a namespace — a plain list comprehension that filters_STATE["pods"].get_logs(pod, ...)looks the pod up, and if it isn't found returns a friendly"Error: pod '...' not found"string instead of crashing. Otherwise it joins the lastlineslog entries with newlines.
What the output means: Every function only reads — it hands back data and changes nothing. get_logs("checkout-api-7d9f"), for example, returns the three log lines including the password authentication failed error.
Try this: Because it's all read-only, this is exactly the surface the read-only diagnostician in Lab 8b needs. Actions that change things live in a separate file (Step 4) so the safety gate can be tested without any risk.
Step 4 · Simulated (fake) actions advanced
The agent will eventually want to do things (restart, scale, open a PR). While building, those are simulated — they return a string saying what they'd do, and change nothing. This lets you exercise the entire safety gate in Lab 8c with zero risk.
mock/actions.pydef rollout_restart(name, namespace="staging"):
return f"[mock] restarted rollout {namespace}/{name}"
def open_pr(title, body, branch):
return f"[mock] opened PR '{title}' from branch '{branch}'"
def delete_pod(name, namespace="staging"):
return f"[mock] DELETED pod {namespace}/{name}" # never actually runs at OBSERVE
# ... scale_deployment, terraform_apply
These are the actions the agent might eventually take — restart a rollout, open a pull request, delete a pod. While you're building, they are simulated: each just returns a string describing what it would do and changes nothing at all.
- Every function returns an f-string like
f"[mock] restarted rollout {namespace}/{name}". The[mock]prefix is a constant reminder that nothing real happened. - The functions span the risk range on purpose:
rollout_restartandopen_prare reversible, whiledelete_podis irreversible — the dangerous kind the safety gate must guard. - Because they only return text, you can wire them into the full safety gate in Lab 8c and watch it allow, block, or ask-for-approval on each one — with zero chance of touching real infrastructure.
What the output means: Calling any of these just gives you a [mock] ... sentence. delete_pod("x") returns "[mock] DELETED pod staging/x" and the real cluster is untouched.
Try this: Splitting reads (cluster.py) from writes (actions.py) is the key design move: it means the diagnostician literally has no function that can change anything, which is a much stronger safety guarantee than 'we promise not to'.
Step 5 · Write and run your first tests expert
Before any API call, prove the foundation works. These tests need no API key — they check the mock and (in Lab 8c) the safety logic directly.
tests/test_mock_and_tools.pyimport os, sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from mock import cluster
def test_mock_finds_crashloop_pod():
pod = cluster.get_pod("checkout-api-7d9f")
assert pod is not None
assert pod["status"] == "CrashLoopBackOff"
def test_mock_logs_contain_root_cause():
assert "password authentication failed" in cluster.get_logs("checkout-api-7d9f")
def test_missing_pod_is_handled():
assert cluster.get_logs("does-not-exist").startswith("Error")
Run them:
terminalcd devops-agent
python -m pytest tests/ -v
tests/test_mock_and_tools.py::test_mock_finds_crashloop_pod PASSED
tests/test_mock_and_tools.py::test_mock_logs_contain_root_cause PASSED
tests/test_mock_and_tools.py::test_missing_pod_is_handled PASSED
3 passed in 0.04s
This is your first test file. Tests are small functions that check your code does what you expect, automatically. These need no API key and cost nothing — they just poke the mock cluster and assert the answers are right.
- The two
sys.path.insert(...)/from mock import clusterlines at the top make themockpackage importable when pytest runs the file. - Each
def test_...()is one test. pytest finds every function whose name starts withtest_and runs it. Inside,assertchecks a condition — if it's false, the test fails and pytest tells you exactly where. test_mock_finds_crashloop_podproves the fixtures load and lookup works.test_mock_logs_contain_root_causeproves the real error is present in the logs (so the agent will have genuine evidence).test_missing_pod_is_handledasks for"does-not-exist"and checks the reply starts with"Error"— proving bad input doesn't crash anything.- You run them with
python -m pytest tests/ -vfrom insidedevops-agent.
What the output means: pytest prints one line per test with PASSED, then a summary: 3 passed in 0.04s. Green here means the foundation is solid — with no API key and no cost. That's the mock-first payoff.
Try this: Break a test on purpose: change "CrashLoopBackOff" in the assert to "Running" and re-run. pytest will show a FAILED with the expected-vs-actual values — that's the feedback loop tests give you.
| Test | Proves |
|---|---|
| crashloop pod is found | Fixtures load; lookup works |
| logs contain the root cause | The agent will have real evidence to find |
| missing pod handled gracefully | No crash on bad input — returns an error string |
Troubleshooting expert
| Symptom | Cause | Fix |
|---|---|---|
ModuleNotFoundError: No module named 'mock' | Running from the wrong directory, or missing __init__.py | cd devops-agent first; ensure mock/__init__.py exists |
No module named pytest | venv not active, or pytest not installed | source .venv/bin/activate then pip install pytest |
No module named pydantic | Dependencies not installed in this venv | pip install -r ../requirements.txt |
FileNotFoundError: fixtures.json | Path resolves relative to CWD, not the module | The provided cluster.py uses os.path.dirname(__file__) — use that pattern, don't hardcode |
| Tests collected: 0 | Test files/functions not named test_* | Prefix files with test_ and functions with test_ |
🪜 Practice ladder beginner → industry
Six graded exercises, easy to real-world. Try each before opening its solution.
Context: Every capstone starts with a clean, importable skeleton — the difference between code that runs offline on day one and a session lost to ModuleNotFoundError.
Your task: Write the terminal commands from Lab 8a Step 1 that create the devops-agent/ project with its five sub-folders and make agent and mock importable Python packages.
Requirements:
- Create the project and its five sub-folders (one
mkdir, brace expansion is fine) - Install
pytestso the offline tests can run - Add
__init__.pytoagentandmockso they import as packages - Explain that the missing
__init__.pyis what causesModuleNotFoundError - Everything runs offline with no API key
💡 Hint: The two __init__.py files are the load-bearing part — without them the imports later in the lab silently fail.
Show solution
cd llm-course-starter
source .venv/bin/activate # same venv as chapters 1-6
pip install pytest
mkdir -p devops-agent/{agent,mock,runbooks,tests,evals}
cd devops-agent
touch agent/__init__.py mock/__init__.py # make them importable packages
The {agent,mock,runbooks,tests,evals} brace-expansion creates all five sub-folders in one mkdir. The two touch'd __init__.py files are what turn agent/ and mock/ into real packages, so later from mock import cluster works. Without them you get ModuleNotFoundError: No module named 'mock'. This step runs entirely offline with no API key.
Context: The safety gate keys every decision on an operation's risk, so the risk levels themselves must be a fixed, typo-proof, JSON-friendly set — not free-form strings.
Your task: Reproduce the RiskClass enum from agent/schemas.py exactly, and explain why it subclasses both str and Enum.
Requirements:
- Four members:
READ_ONLY,REVERSIBLE,SIGNIFICANT,IRREVERSIBLE - Subclass
strso members serialise cleanly to JSON - Subclass
Enumso the set is fixed and a misspelling can't sneak in - Explain how this prevents a typo like "irreversable" ever reaching the gate
- Pure definition — no API key
💡 Hint: A plain string field would let a typo become a brand-new, ungated risk level; the enum makes the set closed.
Show solution
from enum import Enum
class RiskClass(str, Enum):
READ_ONLY = "read_only" # 🟢 inspect only
REVERSIBLE = "reversible" # 🟡 scale, restart, open PR
SIGNIFICANT = "significant" # 🟠 apply non-prod, merge PR
IRREVERSIBLE = "irreversible" # 🔴 delete, prod apply
These four values are the spine of all safety: every tool is later tagged with exactly one of them, and the safety gate (Lab 8c) does comparisons like if risk == RiskClass.IRREVERSIBLE. Inheriting from str means each member is also a real string (so RiskClass.READ_ONLY.value == "read_only" and it serializes cleanly to JSON), while Enum makes the set fixed and typo-proof — you can never accidentally write "irreversable" and slip past the gate. This is a pure definition; nothing runs and no API key is needed.
Context: The agent's diagnosis is structured data the rest of the system trusts, so it needs a schema that validates its own fields — especially a confidence that can't drift outside [0, 1].
Your task: Write the Diagnosis Pydantic model from agent/schemas.py, including a fix_risk: RiskClass field and a bounded confidence, then construct one valid instance.
Requirements:
- Fields for likely cause, a list of evidence, and the suggested fix
- A
fix_riskfield typed asRiskClass confidenceconstrained to [0.0, 1.0] with a PydanticField- An optional runbook reference defaulting to
None - Constructing an out-of-range confidence raises a
ValidationError
💡 Hint: Let the type system do the checking — a Field(ge=0.0, le=1.0) means you never hand-write a range check.
Show solution
from typing import Optional
from pydantic import BaseModel, Field
from agent.schemas import RiskClass
class Diagnosis(BaseModel):
likely_cause: str
evidence: list[str] # the facts it's citing
suggested_fix: str
fix_risk: RiskClass # how risky the fix is
confidence: float = Field(ge=0.0, le=1.0)
runbook_ref: Optional[str] = None
d = Diagnosis(
likely_cause="Broken DATABASE_URL secret from a recent deploy",
evidence=["password authentication failed for user 'checkout'",
"deploy 18 min ago changed DATABASE_URL secret ref"],
suggested_fix="Correct the secret and roll the deployment via a PR",
fix_risk=RiskClass.REVERSIBLE,
confidence=0.9,
)
print(d.fix_risk, d.confidence)
Pydantic validates every field: fix_risk must be a valid RiskClass member and Field(ge=0.0, le=1.0) forces confidence into the closed interval [0, 1] — passing confidence=1.5 raises a ValidationError before the bad data can spread. runbook_ref is optional and defaults to None. This is runnable pure Python (requires only pydantic, no API key).
Context: You develop the agent against a mock cluster so tests run offline and free. The read accessors must load fixtures relative to their own module and degrade gracefully on a missing pod.
Your task: Implement mock/cluster.py's module-level state load plus get_pod, get_pods and get_logs, matching the lesson.
Requirements:
- Locate
fixtures.jsonrelative to the module file, not the current directory - Load the fixture state once at import
get_pods(namespace)returns the pods in that namespace;get_podreturns one orNoneget_logsreturns a friendly error string for a missing pod, else the last N log lines- Pure Python against the fixture — no API key
💡 Hint: Deriving the fixtures path from os.path.dirname(__file__) is what makes the mock work no matter where pytest is launched from.
Show solution
import json, os
_STATE = json.load(open(os.path.join(os.path.dirname(__file__), "fixtures.json")))
def get_pods(namespace="staging"):
return [p for p in _STATE["pods"] if p["namespace"] == namespace]
def get_pod(name, namespace="staging"):
return next((p for p in get_pods(namespace) if p["name"] == name), None)
def get_logs(pod, namespace="staging", lines=20):
p = get_pod(pod, namespace)
if not p:
return f"Error: pod '{pod}' not found"
return "\n".join(p.get("logs", [])[-lines:])
Using os.path.dirname(__file__) resolves fixtures.json next to the module, so it loads no matter which directory you run pytest from — hardcoding a relative path is the usual cause of FileNotFoundError. Every function only reads: get_logs looks the pod up and returns "Error: pod '...' not found" instead of crashing on bad input. This is pure Python and needs no API key (a valid fixtures.json alongside it is required to run).
Context: Before wiring in a model, you prove the mock behaves — the crash-loop is findable, the root cause is in the logs, and a missing pod is handled — all for zero cost.
Your task: Write the three offline tests from Lab 8a Step 5 and give the command to run them.
Requirements:
- Test that the crash-looping pod exists and its status is
CrashLoopBackOff - Test that the root-cause string appears in that pod's logs
- Test that requesting a missing pod returns a graceful error string, not an exception
- Make
mockimportable from the test (e.g. viasys.path) - Give the exact
pytestcommand; all three pass with no API key
💡 Hint: These are assertions against fixed fixture data — if one fails, the mock or the fixture is wrong, and that's exactly what you want to catch before the model is involved.
Show solution
import os, sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from mock import cluster
def test_mock_finds_crashloop_pod():
pod = cluster.get_pod("checkout-api-7d9f")
assert pod is not None
assert pod["status"] == "CrashLoopBackOff"
def test_mock_logs_contain_root_cause():
assert "password authentication failed" in cluster.get_logs("checkout-api-7d9f")
def test_missing_pod_is_handled():
assert cluster.get_logs("does-not-exist").startswith("Error")
cd devops-agent
python -m pytest tests/ -v
# tests/test_mock_and_tools.py::test_mock_finds_crashloop_pod PASSED
# tests/test_mock_and_tools.py::test_mock_logs_contain_root_cause PASSED
# tests/test_mock_and_tools.py::test_missing_pod_is_handled PASSED
# 3 passed in 0.04s
The sys.path.insert(...) line makes the mock package importable when pytest runs the file. pytest auto-discovers every test_* function; each assert proves one property — fixtures load and lookup works, the real error is present so the agent will have genuine evidence, and bad input yields an error string rather than a crash. All three pass with no API key and no cost — the mock-first payoff.
Context: The design splits reads (cluster.py) from simulated writes (actions.py) so safety is enforced by construction: the read module simply has no power to mutate state.
Your task: Implement the simulated actions spanning the risk range and write a test proving cluster.py exposes no function that mutates the fixture state.
Requirements:
actions.pyprovides restart, open-PR and delete that return marker strings without changing state- The actions span the risk range (reversible restart/PR through irreversible delete)
- Snapshot the cluster state (deep copy), call every read accessor, then assert it is unchanged
- The test proves reads are non-mutating by construction
- Pure Python — no API key
💡 Hint: "Safe by construction" beats "safe by review" — if the read functions literally never write, no future edit to them can leak a mutation.
Show solution
# mock/actions.py — every action is SIMULATED (returns text, changes nothing)
def rollout_restart(name, namespace="staging"):
return f"[mock] restarted rollout {namespace}/{name}" # reversible
def open_pr(title, body, branch):
return f"[mock] opened PR '{title}' from branch '{branch}'" # reversible
def delete_pod(name, namespace="staging"):
return f"[mock] DELETED pod {namespace}/{name}" # irreversible
# tests/test_readonly_by_construction.py
import copy
from mock import cluster
def test_reads_never_mutate_state():
before = copy.deepcopy(cluster._STATE)
cluster.get_pods("staging")
cluster.get_logs("checkout-api-7d9f")
cluster.get_pod("checkout-api-7d9f")
assert cluster._STATE == before # nothing a read touched changed state
The [mock] prefix is a constant reminder nothing real happened, and the actions deliberately span the risk range (rollout_restart/open_pr reversible, delete_pod irreversible) so Lab 8c's gate has real targets. The key design move is safety by construction: because reads and writes live in separate modules, the read-only diagnostician literally imports no function that can change anything — a far stronger guarantee than promising not to. The deep-copy test asserts that invariant offline, with no API key.
✓ Checkpoint — Lab 8a complete when…
- Your
devops-agent/skeleton exists with the five packages. RiskClassandDiagnosisare defined and import cleanly.- The mock cluster returns the crash-looping pod and its logs.
python -m pytest tests/ -vshows all green — with no API key.
Knowledge check check yourself
Why is RiskClass defined as an Enum rather than a plain string, given how the safety gate uses it?
Show answer
if risk == RiskClass.IRREVERSIBLE. A typo-proof enum means an action can never accidentally fall through the safety check because someone wrote 'irreversable' — an invalid value is impossible, so a mistagged action can't silently bypass the gate.The build labs split the mock cluster into read-only accessors (cluster.py) and simulated writes (actions.py). Why is that separation a stronger safety guarantee than just promising not to change things, and why build against a mock at all?