pytest
The problem. An ops script that deletes pods or pages people must be right, and "I ran it once on my laptop" is not a test. pytest is Python's test runner (like Jest); with it you test the logic without a real cluster, then talk to the real one with the Kubernetes client.
What you need to know already: the ops script shape (21.31), venvs (21.29), exit codes (6.5), kubeconfig and ServiceAccounts (15.1, 17.33), pods.json from 7.11.
Files named test_*.py, functions named test_*, plain assert:
# test_health.py
from health import status_to_code
def test_up():
assert status_to_code("UP") == 0
def test_down():
assert status_to_code("DOWN") == 2
(.venv) $ pytest
============================= test session starts ==============================
platform linux -- Python 3.14.4, pytest-9.1.1, pluggy-1.6.0
rootdir: /home/learner/oncall-lab/labs/4a-runtime/python
collected 2 items
test_health.py .F [100%]
=================================== FAILURES ===================================
__________________________________ test_down ___________________________________
def test_down():
> assert status_to_code("DOWN") == 2
E assert 1 == 2
E + where 1 = status_to_code('DOWN')
test_health.py:7: AssertionError
=========================== short test summary info ============================
FAILED test_health.py::test_down - assert 1 == 2
========================= 1 failed, 1 passed in 0.03s ==========================
.passed,Ffailed,Eerror in setup. pytest rewrites asserts so a failure shows the values:assert 1 == 2andwhere 1 = status_to_code('DOWN').- Exit code 0 all passed, 1 some failed, 2 interrupted/collection error, 5 no tests collected - a CI step can rely on it.
pytest -vone line per test,-xstop at the first failure,-k downselect by name,pytest test_health.py::test_downone test.
Fixtures
A fixture is a function that prepares something; a test asks for it by naming it as a parameter (@pytest.fixture is a decorator - a label on the function, like the Java annotations in 21.15):
import pytest
@pytest.fixture
def config_file(tmp_path):
p = tmp_path / "app.conf"
p.write_text("db.pool.max=20\nhttp.client.read.timeout.ms=0\n")
return p
def test_finds_missing_timeout(config_file):
assert find_problems(config_file) == ["http.client.read.timeout.ms is 0 (no timeout)"]
Built in and worth knowing:
tmp_path a fresh temporary directory (pathlib.Path) per test
monkeypatch monkeypatch.setattr(module, "name", fake) / setenv / delenv - undone after the test
capsys capture stdout/stderr: out, err = capsys.readouterr()
monkeypatch is how you test code that calls the network without a network (a lambda is Python's short anonymous function, like (url, timeout) => "DOWN"):
def test_down_service_is_exit_2(monkeypatch):
monkeypatch.setattr(health, "fetch_status", lambda url, timeout: "DOWN")
assert health.main(["--url", "http://x"]) == 2
Shared fixtures live in conftest.py in the test directory.
The Kubernetes client
The Kubernetes Python client (pip install kubernetes) calls the same API kubectl does, and returns Python objects instead of text:
from kubernetes import client, config
config.load_kube_config() # ~/.kube/config (what kubectl uses)
# config.load_incluster_config() # inside a pod: the ServiceAccount token
v1 = client.CoreV1Api()
for pod in v1.list_pod_for_all_namespaces().items:
print(pod.metadata.namespace, pod.metadata.name, pod.status.phase)
for c in pod.spec.containers:
limits = c.resources.limits or {} # None when unset
if "memory" not in limits:
print(" no memory limit:", c.name)
- Objects mirror the API:
pod.metadata.name,pod.spec.containers,c.resources.limits- Python attribute names are snake_case (restart_count,container_statuses). - A missing config raises
kubernetes.config.config_exception.ConfigException: Invalid kube-config file. No configuration found. - In a pod, use
load_incluster_config()and give the ServiceAccount a Role that allows exactly what the tool needs (list pods), nothing more.
(simulator) On this box list_pod_for_all_namespaces() and list_namespaced_pod() are backed by the ~/labs/data/pods.json fixture from chapter 7, and need a ~/.kube/config file to exist. Other APIs raise a labelled NotImplementedError.
Later (Ch 22): the cloud SDKs (azure-identity, azure-mgmt-*) work the same way as the Kubernetes client, with a credential object in place of the kubeconfig.
What you can now do
- Write pytest tests, read a failure, and use fixtures (
tmp_path,monkeypatch) to avoid the network. - List and inspect pods from Python with the Kubernetes client, from a kubeconfig or inside a pod.