What you'll learn
Your first four Applied Python lessons gave you project structure, CLI configuration, validated files and a read-only API client. Now we make them recoverable, observable and testable. A reliable utility doesn't just work—it can explain how it behaved when something went wrong.
- Define success, invalid and incomplete outcomes before implementing retries or publishing output.
- Use named Python loggers, severity levels, handlers and stderr diagnostics without duplicate records.
- Record correlation IDs and allowlisted structured events without exposing credentials, tokens or raw payloads.
- Catch precise exceptions and preserve failure meaning instead of converting errors to empty success.
- Write independent unittest test cases with assertions for successful and failed runs.
- Use injected fetch callables, fakes and unittest.mock to reproduce outages without a live service.
- Implement limited retry attempts, controlled wait budgets and final failure handling for read-only requests.
- Produce deterministic, idempotent reports and preserve the previous report when new input is invalid.
- Document a local operator runbook with verifiable test commands and a deliberate readiness gate.
Prerequisite: Complete REST APIs & Authentication and Filesystem & Structured Data Pipelines, or be comfortable with Python functions, JSON schemas, errors and injected API fixtures. The next stage will tackle scheduled execution; this one proves that the code is safe enough to consider scheduling.
Define reliability before adding logging
A program that prints a report once is not automatically reliable. Before writing recovery code, decide what success means: the input meets a known schema, all required pages are complete, the output is deterministic, and the process exits with a status another tool can interpret. An incomplete inventory must not be presented as an empty, healthy inventory. Record which failures are recoverable, which must stop immediately, and what evidence a technician needs to diagnose them. A controlled failure is better than a misleading success.
Keep inputs and outputs explicit
Make the report function accept data rather than reading a global variable or remote endpoint unexpectedly. Define accepted records, rejected records, completion state and the output schema as part of its contract.
Three different outcomes
Valid empty means the schema passed and there really are no records. Invalid input means the schema was not met. Incomplete means required work stopped before collecting all evidence. Do not return an indistinguishable [] for all three.
Reliability includes non-effects
If an API page fails, the program should not replace a prior report, make an unauthorized change or silently claim success. Safety depends on what your tool deliberately does not do.
def validate_inventory(payload):
if not isinstance(payload, dict) or not isinstance(payload.get('devices'),list):
return {'ok':False,'reason':'invalid-schema'}
return {'ok':True,'count':len(payload['devices'])}
print(validate_inventory({'devices':[]}))Why can a program that always returns an empty list after errors be more dangerous than one that raises a clear error? Reveal the explanation after predicting.
A caller could interpret the empty list as a successful inventory containing zero devices, then publish a false report or take a wrong operational action. Returning an explicit invalid or incomplete outcome preserves the distinction between a healthy empty dataset and missing evidence.
Use the logging module instead of scattered print statements
Python's logging module emits records with severity, origin, message and optional exception information. DEBUG supports diagnosis, INFO records normal milestones, WARNING describes unexpected but potentially recoverable conditions, ERROR describes work that failed, and CRITICAL is for severe failures. Use logging.getLogger(__name__) in reusable modules, and configure handlers once at the application boundary rather than in every imported function. The root logger and child loggers can propagate records, so attaching duplicate handlers creates duplicate output. The CLI can send diagnostics to stderr while reserving stdout for machine-readable JSON.
Level is a decision, not decoration
Use INFO for a completed page count and WARNING for a bounded retry. Use ERROR when the report cannot complete. Avoid treating routine successful operation as an error just because it deserves attention.
Logging is configurable
logging.basicConfig configures the root logging setup when it has no handlers; repeated calls in a running interpreter may not reset it. For reusable libraries, create named loggers and let the application configure handlers.
Messages should be useful under pressure
A message such as 'request failed' lacks context. A structured event name plus status, attempt count and nonsecret request ID makes a ticket actionable without exposing credentials or customer data.
import logging
logger=logging.getLogger('northstar.inventory')
logger.addHandler(logging.NullHandler())
logger.info('inventory.page_complete', extra={'page':1})
print(logger.name)Why might a log event appear twice if a developer attaches one handler to the module logger and another to its parent? Reveal the explanation after predicting.
Records normally propagate from child loggers to ancestor handlers. Adding handlers at multiple levels can process the same record twice. Configure handlers centrally at the application boundary, and use NullHandler in a reusable library when appropriate.
Construct and remove a temporary log handler
Before the log-capture challenge, learners need to see the handler, formatter, level, propagation and cleanup operations in one runnable example. This is real Python, not a simulated visualization.
import logging
from io import StringIO
stream = StringIO()
logger = logging.getLogger('northstar.practice')
logger.setLevel(logging.INFO)
logger.propagate = False
handler = logging.StreamHandler(stream)
handler.setFormatter(logging.Formatter('%(levelname)s:%(message)s'))
logger.addHandler(handler)
try:
logger.info('inventory.complete')
finally:
logger.removeHandler(handler)
handler.close()
print(stream.getvalue().strip())StreamHandler chooses where records go; Formatter defines how they look; setLevel determines which records are eligible. propagate=False prevents records from also reaching parent handlers. The finally block removes the handler even if work fails, so repeated calls do not accumulate duplicate output.
Predict and explain: If you attach a handler inside a reusable function but never remove it, what could happen on the second call?
The same event can be emitted more than once because both the old and new handlers remain attached.
Add correlation IDs and redact sensitive context
For a scheduled inventory process, record enough nonsecret context to reconstruct a run: an internal run ID, operation name, page number, attempt count and final outcome. Correlation IDs let operators connect related events without logging whole API bodies. Use a controlled set of allowed fields and sanitize user-supplied values before they reach the logging system. Redacting only the Authorization header is not enough if tokens also occur inside URLs, messages or exception text. A narrow event allowlist is easier to reason about than logging every object and hoping a regex finds secrets.
One run, multiple events
Use the same safe run_id for start, page, retry and completion events. Avoid using an access token, email address or hostname as the correlation ID.
Allowlist fields rather than dump raw state
An event can include status=429, page=2 and attempt=1. It should not dump requests, headers, full exception repr(), sensitive inventory rows or environment variables.
Stable structured event names
Names such as inventory.started, inventory.page_failed and inventory.completed are easier to search than a new sentence for every branch. Store machine-safe fields alongside the event in an approved log sink.
ALLOWED={'event','run_id','status','attempt'}
def safe_event(values):
return {key:values[key] for key in ALLOWED if key in values}
print(safe_event({'event':'inventory.retry','attempt':2,'token':'SECRET'}))If a request fails with 401, should the log include the entire request object to help debugging? Reveal the explanation after predicting.
No. A raw request often contains Authorization headers, tokens, cookies or credential-bearing URLs. Record a safe event name, status, request identifier and attempt number instead. A known set of approved fields supplies useful evidence without copying secrets.
Handle exceptions without swallowing failures
Exceptions help a program distinguish expected failure modes from bugs. Catch specific errors at the boundary that can respond meaningfully: FileNotFoundError when an input is missing, json.JSONDecodeError when syntax is wrong, TimeoutError for a timed-out fetch, and a domain-specific ValueError for invalid schema. Catching Exception and returning [] everywhere can hide programming mistakes and generate false reports. Use raise ... from ... to preserve a useful cause when translating low-level errors; log the sanitized category and return a nonzero CLI status if the run cannot finish. A finally block is for cleanup, not a place to override errors with return.
Catch only what you can act on
If a specific retryable transport error occurs, a bounded retry may be appropriate. If an API returns 403, the script must stop instead of treating it as another timeout.
Errors need an owner
A pure parser may raise ValueError('invalid-schema'); main(argv) can display a concise diagnostic and return status 1. The parser should not print success or replace files while recovering.
Preserve the cause without leaking secrets
Use exception chaining for internal tests, but sanitize user-facing logs. Raw URLs, request headers and stack-trace locals can inadvertently contain sensitive information.
import json
def read_payload(text):
try:
result=json.loads(text)
except json.JSONDecodeError as error:
raise ValueError('invalid-json') from error
if not isinstance(result,dict):
raise ValueError('invalid-schema')
return result
print(read_payload('{}'))Why should a malformed JSON document raise an explicit error instead of producing an empty inventory? Reveal the explanation after predicting.
Malformed JSON is evidence that parsing failed; it says nothing about the number or health of devices. An explicit error lets the CLI stop, preserve the last reliable report, and record what stage requires investigation.
Write meaningful unit tests with unittest
The standard-library unittest framework organizes independent tests in classes derived from unittest.TestCase. Tests use assertion methods such as assertEqual, assertTrue, assertRaises and assertLogs. A good test includes a known input, a specific expected outcome and a clear failure when behavior changes. Check more than the happy path: valid empty inputs, malformed schema, unauthorized responses, duplicate records and failure-without-write. Running `python -m unittest discover -v` in a local project finds test_*.py files by default and provides a reliable pass/fail exit status. A green test suite proves only the cases and assumptions you actually wrote.
Arrange, act, assert
Arrange a fictional input, call the pure function under test, then assert its exact returned value. Avoid using only assertTrue(result), which may pass even when wrong output is merely nonempty.
Test errors as contracts
Use with self.assertRaises(ValueError) to verify malformed input fails. Verify that the operation leaves existing report data unchanged after a rejected input.
Tests run without a UI
A clean machine with the documented project environment can run python -m unittest -v. The browser workbench can execute standard-library tests too, but it cannot verify your actual local installation.
import unittest
def count_hosts(rows): return len(rows)
class CountTests(unittest.TestCase):
def test_empty(self): self.assertEqual(count_hosts([]),0)
case=CountTests('test_empty')
result=unittest.TestResult()
case.run(result)
print(result.wasSuccessful())Is it enough to test only that a normal inventory sample returns two devices, with no failure cases? Reveal the explanation after predicting.
No. That proves the function handles one fixture. Also test valid empty data, malformed input, duplicates and changes in ordering, plus failure modes such as timeouts. Verify that a failed run never publishes or replaces a report.
What class, self and TestCase actually mean
Python Fundamentals taught functions, not object-oriented class syntax. unittest introduces a small amount of that syntax; teach the pieces needed before asking learners to author test methods.
import unittest
def normalize(value):
if value not in ('online', 'offline'):
raise ValueError('invalid-status')
return value
class StatusTests(unittest.TestCase):
def test_online(self):
self.assertEqual(normalize('online'), 'online')
def test_invalid(self):
with self.assertRaises(ValueError):
normalize('unknown')
suite = unittest.defaultTestLoader.loadTestsFromTestCase(StatusTests)
result = unittest.TestResult()
suite.run(result)
print(result.testsRun, result.wasSuccessful())class creates a test class, and (unittest.TestCase) makes it inherit assertion helpers. self refers to the particular test instance; methods named test_* are discovered as tests. assertEqual compares actual/expected results, while assertRaises expects a specific failure. This small amount of class syntax is enough here; no separate OOP course is assumed.
Predict and explain: Why is `self.assertEqual(...)` available inside a class derived from unittest.TestCase?
TestCase supplies assertion methods and self refers to the current instance of that test class.
Use dependency injection and mocks to reproduce failures
Production API behavior is nondeterministic: services throttle, connections time out, and datasets change. For tests, pass a fetch(url) callable into the reporting function instead of constructing a network client deep inside it. The unittest.mock library provides Mock and side_effect to simulate several responses or errors across calls; a tiny handwritten fake dictionary is often sufficient for beginner-level cases. Dependency injection keeps business logic testable without requesting real tokens or contacting external APIs. If you use patch(), patch the name where the code under test looks it up, not necessarily where the dependency was originally defined.
One function, several environments
A function that accepts fetch can use a fake in tests, a local loopback adapter in training, and an approved HTTPS adapter in production without rewriting validation logic.
Mock calls, not success claims
With Mock(side_effect=[TimeoutError(), {'status':200,...}]), the first call raises and the second succeeds. Your retry code must decide whether another attempt is permitted and count calls exactly.
No hidden global state
Reset mocks and fixtures between tests so one test doesn't depend on the last run's counter, cached result, logger handlers or temporary files.
from unittest.mock import Mock
fake=Mock(side_effect=[TimeoutError('demo'),{'status':200}])
try: fake()
except TimeoutError: pass
print(fake()['status'],fake.call_count)Why is an injected fake HTTP function preferable to running automated tests against a live tenant? Reveal the explanation after predicting.
The fake makes response order, timing and failures repeatable without secrets or network dependencies. You can assert exact calls and outputs, including failures that would be risky or unreliable to reproduce in a real environment.
Lambda is a small function, not a mysterious new API
The next lessons use injected callbacks such as lambda: response and lambda page: pages[page]. They are anonymous functions: a concise alternative to writing def for a one-expression function.
def get_static_response():
return {'status': 200}
short_response = lambda: {'status': 200}
pages = {1: {'status': 200}}
fetch_page = lambda number: pages[number]
print(get_static_response()['status'])
print(short_response()['status'])
print(fetch_page(1)['status'])lambda: expression accepts no arguments. lambda number: expression accepts one argument. The body is one expression, and the function is called later. Prefer def for multi-step logic, validation, exception handling and production clarity.
Predict and explain: Does `lambda page: pages[page]` fetch a live API page?
No. It returns a value from a local dictionary when invoked with a page number. The caller controls where the data comes from.
Bound retries, backoff and the overall work budget
Not every failure is worth retrying. Timeouts, some 5xx responses and 429 throttling may be temporary for a safe read-only GET, while 400, 401, 403 and invalid JSON usually need correction rather than repetition. Limit attempts explicitly, respect valid Retry-After within your allowed deadline, and apply capped backoff with jitter when appropriate to reduce synchronized load. A production client needs both per-request timeouts and an overall deadline; simply setting max_attempts is not a time budget. Tests should inject a sleep callback and a clock instead of actually delaying so retry logic can be evaluated quickly and deterministically.
Attempt count includes the first try
max_attempts=3 means at most three calls, not three retries after the first. Tests should fail if an implementation accidentally makes four calls.
Do not swallow the final failure
When the retry budget is exhausted, return an incomplete result or raise a documented exception. Do not replace the last good report with partial data.
Respect method semantics
Retrying GET is often safer than replaying POST. A write may have succeeded even if its response was lost; only retry state changes with a documented idempotency mechanism and authorization.
def attempts(call, maximum):
for number in range(1,maximum+1):
try: return {'ok':True,'value':call(),'attempts':number}
except TimeoutError:
if number==maximum: return {'ok':False,'reason':'timeout','attempts':number}
print(attempts(lambda:42,3))Why can't a retry policy simply keep calling an endpoint until it returns HTTP 200? Reveal the explanation after predicting.
Unbounded retries can exhaust quotas, hold scheduled jobs indefinitely and repeatedly invoke unsafe state changes. The client needs an attempt and time budget, specific retryable errors, capped waits and a final explicit failure when the budget is exhausted.
Will the automation publish a trustworthy report?
Choose an incident, adjust the retry budget, and inspect the generated events. The model uses fictional API responses: it does not execute Python, contact any service, save files, or collect credentials. Real Python exercises below test the same policies using mocks.
{"run_id":"demo-run-01","attempt_budget":2}{"run_id":"demo-run-01","attempt":1}{"run_id":"demo-run-01","attempt":1,"reason":"timeout"}{"run_id":"demo-run-01","attempt":2}{"run_id":"demo-run-01","attempt":2,"total":1}The first attempt times out, but a second approved GET succeeds within the attempt budget.
Previous known-good report: 7 devices. Any failed run leaves that result untouched and returns an explicit reason.
Teaching simplification: the explorer models attempt decisions only, not elapsed deadlines or actual waiting. A real rate-limited client needs a validated Retry-After policy, per-request timeouts, and an overall execution budget.
Make repeated runs predictable and protect known-good output
Idempotency means repeating an operation with the same inputs does not keep changing its intended result. A read-only inventory report should sort records and produce stable JSON, not append duplicate lines with each run. When publishing a file, validate all input first, write to a temporary file in the destination directory, and replace the destination only when authorized and the complete report is ready. Logging timestamps or run IDs belong in a separate execution log when exact report bytes need to remain deterministic. A local os.replace helps with atomic visibility on supported filesystems but is not crash-durable by itself or a complete safeguard for adversarial shared directories.
Check before commit
A successful GET for page one is not enough. Complete all pages and schema checks before preparing a report candidate; on any later failure, leave the previous report intact.
Run twice against the same fixtures
Repeat the full transformation and compare exact report bytes or parsed data. The same input should produce the same report order and counts.
Explicit change boundary
Dry-run shows what would be written. Apply requires a separate approved action, and the program must not mix uncontrolled side effects into a supposedly read-only preview.
import json
def deterministic_report(hosts):
return json.dumps({'hosts':sorted(set(hosts))},sort_keys=True)
first=deterministic_report(['SW-02','AP-01'])
second=deterministic_report(['AP-01','SW-02'])
print(first==second)What should happen to a previously valid inventory report if the latest API fetch times out on page two? Reveal the explanation after predicting.
The new run must report an incomplete result and leave the previously valid report untouched. It cannot claim zero devices or publish the successful first page as the full inventory, even if part of the new data looks correct.
Ship an observable tool with tests and an operator runbook
An automation is more useful when someone else can run it, understand its expected behavior and troubleshoot failures without reading every line of source. Document supported Python versions, setup commands, input and output locations, permissions required, typical logging events, exit codes and recovery steps. Add repeatable unittest cases alongside the implementation and show the exact command to run them. In the local project below, you'll extend the fictional inventory reporter with a named logger, a fake API responder, a bounded fetch policy, deterministic report output and a test file that validates successful, timed-out and unauthorized outcomes. No real API tokens or customer data are required.
Operations handoff must answer why
Describe the difference between 0 success and nonzero failure, the retry ceiling, what the log event tells the technician, and why a failed run does not overwrite the prior report.
Verify the failure as carefully as success
Run a test where the first attempt times out but the second succeeds, then a test where every attempt times out. Confirm the exact call count and failure category.
An explicit readiness gate
Before scheduling, require passing tests, harmless default behavior, clean failure messages, a documented runbook and a reviewed execution identity. Scheduled execution comes in Lesson 06, not by assumption.
def run_inventory(fetch):
try: return {'ok':True,'payload':fetch()}
except TimeoutError: return {'ok':False,'reason':'timeout'}
print(run_inventory(lambda:{'devices':[]}))What would you want to see before allowing a report tool to run unattended every morning? Reveal the explanation after predicting.
A documented environment and least-privilege identity, tested normal and failure cases, clear diagnostics and exit codes, bounded retries and timeouts, deterministic output, a safe default preview, preservation of known-good reports and an operator recovery procedure.
Run an inventory reporter and its tests on your computer
Inside the local project from Lesson 01, save the following source as reliable_inventory.py. It operates only on fictional in-memory API responses, never makes live requests, and does not modify files. The module provides validation, named logging, a two-attempt transient-failure budget and a stable report result. The CLI uses stdout for a completed JSON report and stderr for errors.
Evidence before automation
Run the script with success, recover, timeout, auth and invalid response scenarios. Verify that the output is never presented as a complete inventory when the underlying data is incomplete, unauthorized or invalid.
Do not add live credentials
This lesson requires no network connectivity, keys or Microsoft tenant. Logs record a fictional event name and limited metadata, never a real access token. Run local commands from your own practice folder, not a managed customer directory.
"""Northstar Services — reliable, read-only inventory demo (fictional data only)."""
import argparse
import json
import logging
import sys
LOG = logging.getLogger("northstar.reliability")
PREVIOUS = {"total": 7, "offline_hosts": ["OLD-01"]}
def report_from_response(reply):
"""Validate a complete response before creating any report."""
if not isinstance(reply, dict) or reply.get("status") != 200:
raise ValueError("invalid-response")
body = reply.get("body")
if not isinstance(body, dict) or not isinstance(body.get("devices"), list):
raise ValueError("invalid-schema")
devices = body["devices"]
names = []
offline = []
for item in devices:
if not isinstance(item, dict):
raise ValueError("invalid-device")
host = item.get("host")
status = item.get("status")
if (not isinstance(host, str) or not host.strip()
or not isinstance(status, str)
or status.strip().lower() not in {"online", "offline"}):
raise ValueError("invalid-device")
name = host.strip().upper()
names.append(name)
if status.strip().lower() == "offline":
offline.append(name)
if len(names) != len(set(names)):
raise ValueError("duplicate-device")
return {"total": len(names), "offline_hosts": sorted(offline)}
def run_inventory(fetch, previous, max_attempts=2):
"""Retry only a read-only, transient failure; never publish partial data."""
if type(max_attempts) is not int or max_attempts < 1:
return {"ok": False, "previous": previous, "error": "invalid-attempts", "attempts": 0}
LOG.info("inventory.started")
for attempt in range(1, max_attempts + 1):
try:
reply = fetch()
except TimeoutError:
LOG.warning("inventory.transient_failure attempt=%d kind=timeout", attempt)
if attempt == max_attempts:
LOG.error("inventory.incomplete reason=timeout")
return {"ok": False, "previous": previous, "error": "timeout", "attempts": attempt}
continue
if not isinstance(reply, dict) or type(reply.get("status")) is not int:
LOG.error("inventory.incomplete reason=invalid-envelope")
return {"ok": False, "previous": previous, "error": "invalid-envelope", "attempts": attempt}
if reply["status"] == 503:
LOG.warning("inventory.transient_failure attempt=%d status=503", attempt)
if attempt == max_attempts:
return {"ok": False, "previous": previous, "error": "http-503", "attempts": attempt}
continue
if reply["status"] != 200:
LOG.error("inventory.incomplete status=%d", reply["status"])
return {"ok": False, "previous": previous, "error": f"http-{reply['status']}", "attempts": attempt}
try:
report = report_from_response(reply)
except ValueError as exc:
LOG.error("inventory.incomplete category=%s", str(exc))
return {"ok": False, "previous": previous, "error": "invalid", "attempts": attempt}
LOG.info("inventory.completed attempts=%d total=%d", attempt, report["total"])
return {"ok": True, "report": report, "attempts": attempt}
raise AssertionError("unreachable retry state")
def build_demo_fetch(scenario):
good = {"status": 200, "body": {"devices": [
{"host": "SW-01", "status": "online"},
{"host": "AP-02", "status": "offline"}]}}
if scenario == "success":
responses = [good]
elif scenario == "recover":
responses = [TimeoutError("fictional timeout"), good]
elif scenario == "timeout":
responses = [TimeoutError("fictional timeout")] * 3
elif scenario == "auth":
responses = [{"status": 403}]
else:
responses = [{"status": 200, "body": {"devices": [{"host": "", "status": "online"}]}}]
iterator = iter(responses)
def fetch():
result = next(iterator)
if isinstance(result, Exception):
raise result
return result
return fetch
def main(argv=None):
parser = argparse.ArgumentParser(description="Offline reliability exercise; no network or writes")
parser.add_argument("--scenario", choices=["success", "recover", "timeout", "auth", "invalid"], default="success")
parser.add_argument("--verbose", action="store_true")
args = parser.parse_args(argv)
logging.basicConfig(level=logging.INFO if args.verbose else logging.WARNING,
stream=sys.stderr, format="%(levelname)s:%(name)s:%(message)s")
result = run_inventory(build_demo_fetch(args.scenario), PREVIOUS)
if result["ok"]:
print(json.dumps(result["report"], sort_keys=True))
return 0
print("Inventory blocked: " + result["error"], file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())Now save this second file as test_reliable_inventory.py in the same folder. It uses unittest, unittest.mock.Mock and assertLogs to check success, invalid input, deterministic output, exhausted timeouts, permission denials and preservation of known-good data.
"""Independent unittest checks for reliable_inventory.py; no network needed."""
import unittest
from unittest.mock import Mock
from reliable_inventory import PREVIOUS, report_from_response, run_inventory
GOOD = {"status": 200, "body": {"devices": [
{"host": "SW-02", "status": "online"}, {"host": "AP-01", "status": "offline"}]}}
class ReportTests(unittest.TestCase):
def test_deterministic_report(self):
first = report_from_response(GOOD)
changed_order = {"status": 200, "body": {"devices": list(reversed(GOOD["body"]["devices"]))}}
self.assertEqual(first, report_from_response(changed_order))
self.assertEqual(first, {"total": 2, "offline_hosts": ["AP-01"]})
def test_valid_empty_inventory(self):
self.assertEqual(report_from_response({"status": 200, "body": {"devices": []}}),
{"total": 0, "offline_hosts": []})
def test_bad_schema_is_not_empty(self):
with self.assertRaises(ValueError):
report_from_response({"status": 200, "body": {"items": []}})
def test_conflicting_duplicate_hostnames_fail(self):
rows = [{"host": "SW-01", "status": "online"}, {"host": "sw-01", "status": "offline"}]
with self.assertRaises(ValueError):
report_from_response({"status": 200, "body": {"devices": rows}})
class OperationalTests(unittest.TestCase):
def test_transient_timeout_then_success(self):
fetch = Mock(side_effect=[TimeoutError(), GOOD])
outcome = run_inventory(fetch, PREVIOUS)
self.assertTrue(outcome["ok"])
self.assertEqual(outcome["attempts"], 2)
self.assertEqual(fetch.call_count, 2)
def test_exhausted_attempts_preserve_known_good(self):
fetch = Mock(side_effect=TimeoutError())
old = PREVIOUS.copy()
with self.assertLogs("northstar.reliability", level="WARNING") as events:
outcome = run_inventory(fetch, PREVIOUS)
self.assertFalse(outcome["ok"])
self.assertEqual(outcome["error"], "timeout")
self.assertEqual(outcome["previous"], old)
self.assertEqual(PREVIOUS, old)
self.assertEqual(fetch.call_count, 2)
self.assertTrue(any("inventory.incomplete" in line for line in events.output))
def test_permission_failure_is_not_retried(self):
fetch = Mock(return_value={"status": 403})
outcome = run_inventory(fetch, PREVIOUS)
self.assertEqual(outcome["error"], "http-403")
self.assertEqual(fetch.call_count, 1)
def test_bad_payload_preserves_prior_result(self):
outcome = run_inventory(lambda: {"status": 200, "body": {"devices": [None]}}, PREVIOUS)
self.assertEqual(outcome["error"], "invalid")
self.assertEqual(outcome["previous"], PREVIOUS)
if __name__ == "__main__":
unittest.main()Run the normal and failing scenarios
python reliable_inventory.pypython reliable_inventory.py --scenario recover --verbosepython reliable_inventory.py --scenario authpython reliable_inventory.py --scenario timeoutpython reliable_inventory.py --scenario invalidA failed run should return a nonzero process exit status and emit an error to stderr, not a clean-looking JSON inventory to stdout. Check the command's exit status using your local shell.
Run all eight unit tests
python -m unittest test_reliable_inventory.py -vpython -m unittest discover -vAll eight tests should pass on a supported Python 3 environment with no third-party dependencies. Deliberately change an expected value in a test, confirm it fails, then restore it. That is how you verify your tests can detect regressions.
Operating runbook: Document the five CLI scenarios, expected exit codes (0 success; 1 failure), retry limit (two total attempts), log severity conventions, no-write behavior and what the next technician should do after a failed run. Do not schedule this demo: scheduled and controlled execution is Lesson 06.
Practice testing and recovery with real Python
The eight coding challenges cover log severity, safe handler capture, structured event allowlisting, precise parsing failures, actual unittest assertions, injected HTTP fakes, bounded retries and an independent recoverable report. Run your code, observe errors and Check against additional fictional cases—not just the visible example.
Write it. Run it. Understand it.
Implement safe logging, unittest assertions, injected mock failures, bounded retries and deterministic report recovery in real Python.
Choose meaningful logging severity
Write event_level(kind) using Python's logging module, returning the integer logging level for start/complete (INFO), retry (WARNING), and failed (ERROR). Return None for unknown event names. Do not configure handlers or print inside the helper. Print the three level names for the sample. The grader changes event order and checks that a failed operation is not logged as INFO.
INFO,WARNING,ERRORTry solving it before asking for help. Hints never show a completed program.
Your printed output and errors will appear here.
Run the code, then step through the executed lines to watch variable values change.
Prove it behaves safely on the second run.
Try transient failures followed by success, persistent timeouts, authorization failures, invalid JSON and repeated successful input. All incomplete outcomes must preserve previous known-good data.
Continue to all eight reliability challenges →Continue to Lesson 06: Scheduling & Controlled Automation →Return to the Applied Python path →Official references
Use the official Python logging and testing documentation to learn handler propagation, assertion behavior and mocking. More advanced logging sinks and operational retry libraries may behave differently; verify them before adapting a training script for production.