What you'll learn
The first five Applied Python lessons taught you to set up a project, configure a CLI, validate files, consume read-only APIs and test failures. Lesson 06 adds the operational boundary: a scheduler may start a process, but the process must independently prove that its plan is safe and any proposed change is authorized.
- Separate a scheduler trigger from a validated job and separately authorized change.
- Launch the intended Python interpreter and configuration under Windows Task Scheduler, cron or systemd without relying on a terminal session.
- Validate scheduling settings, scope and input types before planning any action.
- Build a deterministic plan and explain why dry-run must suppress every external effect.
- Model approval scope, change windows, expiration and revocation with trusted records.
- Distinguish per-run correlation IDs from durable idempotency keys and replay protection.
- Recognize overlapping execution and implement safe skip/queue policy with real atomic coordination reserved for production.
- Preserve the previous verified result across partial failures and uncertain restart states.
- Write an operator handoff with exit codes, alerts, tests and escalation guidance.
Before beginning: Complete Logging, Testing & Reliability, or know how to write validated functions, use unittest, classify errors and preserve prior results. You do not need Task Scheduler, cron or admin rights to complete the web practice.
Understand the difference between a trigger, a job and an authorized change
A cron entry, Windows Scheduled Task or service timer can start a program, but none of them proves the work was authorized. A trigger answers WHEN to attempt execution; the job describes WHAT to evaluate; an approval gate determines WHETHER a change may occur. Define accepted inputs, read-only behavior, failure statuses and the observable evidence before adding a schedule. Our fictional Northstar Services inventory job normally reads a local snapshot and recommends which hosts deserve investigation. Even after an operator asks to simulate applying its plan, the example never connects to a tenant or modifies devices.
Three separate decisions
A job can be due at 02:00 and still be blocked because its change window is closed or its approval was revoked. Treat due, eligible and authorized as independent checks, and record which one failed.
Default to a safe outcome
An invalid config, empty authorization token or missing snapshot should produce a clear blocked outcome, not an apparently successful empty inventory or an automatic action against infrastructure.
Use a deterministic result contract
Record status, reason, run identifier, number of proposed actions and exit code. Stable input should yield a stable plan, while execution metadata such as attempt time remains separate from the plan itself.
def decide(trigger_due, approved):
if not trigger_due:
return 'not-due'
if not approved:
return 'blocked-approval'
return 'eligible'
print(decide(True, False))If the scheduler starts a valid job at the correct time, why is that not sufficient evidence that the job may change an endpoint or publish a new report? Reveal the explanation after predicting.
Scheduling establishes timing, not authority. The program must separately evaluate whether the requested action fits approved scope, identity, valid window and current safe input. A correctly timed task can still be blocked, and an incomplete run must not masquerade as success.
Schedule the correct interpreter with reliable paths and environment settings
Windows Task Scheduler, cron and systemd timers launch commands in processes with different working directories, environment variables, privileges and time-zone behavior. A script that worked from your interactive terminal can fail unattended because it resolves a different Python executable, cannot find a relative file, or expects an interactive prompt. Use an absolute interpreter path, an explicit working directory and minimally scoped service identity. Test the exact command under an equivalent account before setting recurrence. For this course, copy commands into an isolated practice directory only; the site does not create operating-system scheduled jobs.
Windows scheduled tasks
Configure Program/script as the project virtual environment's python.exe, Arguments as the absolute path to the script plus explicit flags, and Start in as its directory. Use a narrowly scoped account and inspect Task Scheduler history and last-run exit code.
cron and systemd
A cron entry has a small execution environment and may lack your expected PATH. A systemd unit plus timer can express more deliberate service behavior. In both cases document executable, working directory, standard output/error and failure reporting.
Time zones and DST
Specify a local operational time zone and document UTC conversion. A wall-clock occurrence may be skipped or repeated during daylight-saving transitions, and overlapping triggers are not a substitute for a runtime lock or deduplication policy.
from pathlib import Path
import sys
command = [str(Path(sys.executable)), '/training/job.py', '--mode', 'plan']
print(command[-2:])What are two reasons a Python script succeeds in a manually activated virtual environment but fails after Task Scheduler or cron launches it? Reveal the explanation after predicting.
The scheduler may start a different Python interpreter without the intended dependencies, or it may use a different current directory and environment. Explicit executable paths, working directories, configuration and account permissions make unattended behavior reproducible.
A scheduled launch needs an explicit working directory
A scheduler is a process launcher, not an authorization system. Before creating a recurring task, verify the same command manually using an absolute Python interpreter and a disposable local project.
from pathlib import Path
import sys
interpreter = Path(sys.executable)
command = [str(interpreter), '/training/scheduled_job.py', '--mode', 'plan']
print(command[-2:])The command array helps explain what the scheduler should execute, but printing the list does not register a task. The optional local scheduler walkthrough later in this lesson provides actual Windows, cron and systemd configuration.
Predict and explain: Is an hourly trigger the same thing as approval to run a simulated change?
No. Trigger time only indicates when to attempt a run. The job independently validates configuration, scope and approval.
Use explicit configuration precedence and reject unsafe settings
A scheduled script cannot stop to ask a technician what should happen next. Determine configuration precedence and validate the complete effective settings before accessing inputs or performing actions. A defensible pattern is explicit CLI flags over selected environment variables over a validated config file over documented defaults, but the exact order must be defined by your project. Missing, unknown or contradictory values should fail closed. Secrets belong in an approved secret store with appropriately scoped access; never print a whole environment mapping or log configuration values that might include credentials. The offline demo uses no secrets.
Separate the policy from the input
The input dataset tells you what devices exist; configuration tells you which site, maximum actions, mode and schedule window are permitted. A record in an inventory file should not grant itself authorization.
Require bounded types
Validate integers without accepting booleans as counts, restrict allowed modes to plan or simulate-apply, and require nonempty approved site identifiers. Reject unknown keys if the job depends on a strict schema.
Make a failed config obvious
Bad configuration is not a transient HTTP outage. Do not retry invalid values indefinitely or produce a blank report. Record a nonzero exit code and enough safe context to fix the schedule before enabling it again.
def valid_limit(value):
return type(value) is int and 1 <= value <= 25
print(valid_limit(5), valid_limit(True))Why is accepting True as a numeric action limit risky in Python, and how should a job validate that setting? Reveal the explanation after predicting.
bool is a subclass of int in Python, so isinstance(True, int) is True. For a strict bounded count, use type(value) is int and enforce an explicit minimum and maximum. A scheduler should reject malformed configuration before producing a plan or simulated action.
Distinguish read-only planning, dry-run and simulated apply
A plan enumerates proposed operations and the evidence used to justify them. A dry-run should execute the same validation and decision policy as an apply path while suppressing every side effect; simply printing 'DRY RUN' beside a command that still writes a report is not safe. Our local exercise deliberately has only plan and simulate-apply modes, and both are side-effect free. A real production apply would require approved connectors, explicit change windows, permission boundaries, audit logs, rollback procedures and carefully tested external-effect isolation—not copied training flags.
Plan is a first-class artifact
Sort proposed actions deterministically and show the target, operation, reason and action key. Invalid records should block publication of the entire plan instead of silently dropping devices that may matter.
Never trust labels as controls
An --apply or --dry-run string does not enforce safety unless each effectful function checks the selected mode. Keep file writes and API calls behind a narrow, audited boundary, absent from the offline teaching example.
Preview the exact scope
Approval for reviewing a report does not authorize restarting a service, changing a firewall rule or updating identity policy. Make the expected effect and affected resources legible before a person authorizes the change.
def plan_hosts(records):
return sorted(r['host'] for r in records if r['state'] == 'review')
print(plan_hosts([{'host':'SW-02','state':'review'}]))If the --dry-run flag prevents a network change but still overwrites an existing report file, is that a safe read-only preview? Reveal the explanation after predicting.
No. A dry-run that produces a persistent write is still changing external state. Unless a separately approved output is explicitly part of the contract, a true preview should avoid mutation, and tests should inspect every side-effect boundary, including files and API calls.
Should this scheduled job proceed?
Model an unattended Northstar inventory job. Change its inputs, policy and approval state, then inspect the sequence of decisions. This is a deterministic illustration: it does not execute Python, schedule tasks, contact a service, or modify files or devices.
- 01 · Trigger
Scheduler fires a fictional inventory run. Time alone does not authorize an action.
- 02 · Ownership
No overlap reported. In production this would require an atomic shared lock.
- 03 · Input
3 records validated; 2 review actions planned.
- 04 · Policy
Plan mode: present proposed actions, never execute or simulate a change.
read-only plan
Important limitation: approval and overlap controls model policy, not trusted tickets or atomic locks. Real production automation needs independent authorization checks, scoped identities, durable execution history and a verified side-effect boundary.
Check identity, resource scope, expiration and change window
Change control must be validated at the moment of proposed execution. A ticket number typed into a CLI argument is not proof that an approver authorized the operation: it is only an identifier to verify against a trusted approval record. A meaningful gate checks the approved actions and site, the accountable identity, start and end times, and whether approval remains valid. In production, that record must come from the organization’s approved system rather than from the same untrusted input file as the job. Our simulator uses fictional predeclared records, not a real ticketing or identity service.
Scope matters more than presence
An approval for a read-only inventory in Site-A does not cover a configuration change in Site-B. Exact operation, target scope, responsible identity and expected impact must match the plan.
Expiration is a hard boundary
Treat change windows as half-open intervals: start <= current UTC time < end. At the end boundary, even a job that started earlier must not begin a new unauthorized operation.
Revocation and stale plans
An operator may revoke an approval or the inventory may change after review. Record the stable plan digest or independently revalidate scope immediately before the effectful step. A ticket string alone proves neither.
from datetime import datetime, timezone
now=datetime(2026,11,10,2,tzinfo=timezone.utc)
start=datetime(2026,11,10,1,tzinfo=timezone.utc)
end=datetime(2026,11,10,3,tzinfo=timezone.utc)
print(start <= now < end)Why should a job reject an approval record whose ticket ID looks correct but whose permitted site or time window does not match the proposed work? Reveal the explanation after predicting.
Ticket identifiers label approvals but do not establish that every action is in scope. The current request must match the trusted record's site, operation, identity and still-open window. Otherwise an approved change could be replayed against an unintended target or after the authorized maintenance period.
Recognize aware timestamps, offsets and half-open windows
The date-time practice challenges reject naive timestamps. A timestamp without timezone information cannot be compared safely with a separately established UTC maintenance window.
from datetime import datetime, timezone, timedelta
start = datetime(2026, 11, 10, 1, tzinfo=timezone.utc)
end = start + timedelta(hours=2)
now = datetime(2026, 11, 9, 21,
tzinfo=timezone(timedelta(hours=-5)))
naive = datetime(2026, 11, 10, 2)
print(now.utcoffset())
print(start <= now < end)
print(naive.tzinfo is None)
print(naive.utcoffset() is None)tzinfo holds timezone information; utcoffset() is the offset from UTC, or None for a naive datetime. timedelta represents a duration. Different aware offsets can represent the same instant. Compare start <= now < end, which excludes the instant at the end boundary. Require valid datetime instances and aware inputs before comparing.
Predict and explain: Why is the exact end of a 01:00–03:00 UTC change window blocked?
The window is half-open: start is included but end is excluded. At 03:00 UTC the approved interval has expired.
Make reruns idempotent and assign stable action keys
Schedulers repeat launches, operators retry failed runs, and an interrupted job may start again after recovery. Idempotency means repeating the same intended operation does not create additional unintended effects. It is not the same as caching the last exit code, and it does not mean every repeat should be skipped forever. Define stable action identifiers from the operation and normalized target plus an approved change/version key, then compare them against a durable execution ledger in a real implementation. A separate run_id identifies a single invocation for logs; it is not a stable deduplication key.
Stable keys are about intent
A key such as CHG-1042:review:SW-01 describes one approved action. A randomly generated timestamp used as the dedupe key changes every run and cannot prevent duplicate execution.
Persist after a verified effect
For real changes, writing the ledger before or after the effect has tradeoffs if a process crashes between those steps. Prefer idempotent external APIs, transaction semantics or explicit reconciliation; a local set in memory alone does not protect across processes.
Version intentional repeats
New approval scope or a changed desired state may justify a new operation with a distinct key. A repeated identical input should yield the same deterministic plan and should not duplicate work merely because the scheduler fired twice.
def action_key(change, host):
return f'{change}:review:{host.strip().upper()}'
print(action_key('CHG-1042','sw-01'))Why does a fresh run UUID make a poor deduplication key when the same scheduled job is retried after a crash? Reveal the explanation after predicting.
A fresh UUID intentionally identifies the new attempt, so the exact same intended action receives a different value each time. A stable action key must be derived from the approved operation and target; run_id remains useful for tracing one invocation, not for detecting repeat effects.
Prevent overlapping jobs and enforce bounded execution
An hourly scheduler may launch the next run while a delayed prior run is still active. Without a concurrency policy, two workers can read the same old state, generate duplicate plans and attempt conflicting changes. A classroom Boolean 'busy' flag illustrates the decision but is not a cross-process lock. Production-grade coordination needs a shared atomic claim with ownership, lease expiry and a recovery story, or an orchestrator that guarantees the required serialization. Bound maximum run duration and page/attempt budgets so a stuck job does not occupy an execution slot indefinitely.
Overlap is a distinct outcome
Skip, queue or reject a second run deliberately; don't pretend it succeeded. Record which owner has the claim, how the next run behaves and whether skipped work needs an operator follow-up.
Leases can expire
A timed lease can recover after a crash, but an old worker may continue running. Use fencing tokens or a target-side concurrency mechanism where overlapping writes would be dangerous.
Know your time budget
Timeouts should apply to network calls, individual steps and the overall job. A graceful shutdown can preserve safe evidence while a hard stop may require reconciliation. Distinguish elapsed deadlines from the scheduler's next wall-clock trigger.
def start_decision(active, due):
if not due: return 'not-due'
return 'skip-overlap' if active else 'start'
print(start_decision(True, True))If two scheduled processes each check a shared Boolean 'active' and both see False, can either safely claim exclusive ownership? Reveal the explanation after predicting.
No. The check-then-set sequence is not atomic: both processes can observe False before either updates it. A proper inter-process claim requires a trusted atomic mechanism, lease or orchestrator guarantee. The simple Boolean example is only for understanding policy decisions in the browser.
Design failure recovery without publishing partial work
Imagine a scheduled run that validates its configuration, fetches two pages, then crashes just before publication. The next run must determine whether any action occurred and whether a partial result should be discarded or reconciled. The previous verified report remains authoritative until a newly complete report passes validation. Retries should target only transient, bounded failures; never blindly repeat permission denials or unapproved writes. Real systems need durable checkpoints and action receipts, protected from tampering, plus a restart procedure that reconciles uncertain outcomes before resuming execution.
Three useful terminal results
Complete means every required input was validated. Blocked means policy or permissions prevented work. Incomplete means required evidence or execution state is missing; it is not an empty healthy inventory.
Pause on uncertainty
A failure after an external side effect but before the confirmation receipt creates an unknown state. Recheck the target or use an idempotent API before retrying; never assume the action did not occur.
Preserve known-good output
Build the full result in memory or a staged location, verify it, and publish only after success. The classroom model returns a hypothetical publication decision without touching the filesystem.
def publication_state(valid, complete):
return 'publish' if valid and complete else 'preserve-previous'
print(publication_state(True, False))Why is a report containing only the first page of a two-page inventory unsafe to publish as a successful scheduled run? Reveal the explanation after predicting.
The report is incomplete and could wrongly suggest missing devices are absent or healthy. The scheduler should preserve the previous verified report, mark the current run incomplete, emit a nonzero exit status and record sufficient safe evidence for a technician to investigate.
Runbook, monitoring and independent readiness tests
Automation becomes an operational service when another technician can identify its owner, expected schedule, allowed inputs, approved scope, success criteria and failure response. Document the exact Python interpreter, launch directory, configuration sources, time zone, service identity, retry budget, overlap policy, output contract and escalation path. Logs should include a nonsecret run ID and meaningful terminal status, while monitoring alerts on nonzero exits, missed expected runs and unexpected duration. A green scheduler 'launched' indicator is not the same as the job proving it completed correctly.
Evidence for the next technician
Preserve a safe summary of proposed actions, the approval identifier if one exists, completion state and a redacted trace. No real customer hostnames, credentials or raw payloads appear in this training scenario.
Tests must prove blocks
Use unit tests for invalid config, expired approvals, wrong target scope, overlap, duplicate action keys, repeated runs and incomplete data. Also verify the dry-run path produces zero external effects.
Operate through planned change
Before real deployment, obtain platform and security approval, least-privilege identity, protected secrets, tested restore/reconciliation behavior and documented ownership. This course does not grant production permissions or trigger real workflows.
def exit_for(outcome):
return {'complete':0,'blocked':2,'incomplete':3}.get(outcome,4)
print(exit_for('blocked'))How could a scheduler display Last Run Result 0 while operators still miss an incomplete inventory or an unauthorized change? Reveal the explanation after predicting.
Process exit 0 only means the executable reported success. A script that catches errors and returns 0, or a scheduler that launches a wrapper but not its child process correctly, can mask failure. Contract tests, monitored completion evidence, safe logs, exit statuses and authorization audits are all required.
Run a controlled scheduled-job model locally
Create a disposable practice folder, ideally in the virtual environment from Lesson 01. Save the following source as scheduled_job.py. It accepts a mode and a fictional incident scenario, validates the entire input, constructs an ordered plan, checks a hardcoded training approval record, and returns explicit blocked, planned or simulated results. No network requests, service processes, scheduled tasks, live ticket lookups or writes are performed.
Plan is the default
The default invocation is read-only. It describes proposed reviews and returns the old verified report unchanged. An unknown or invalid input blocks, even when the scheduler fires at the correct time.
Simulation is not change authorization
Even with --mode simulate-apply and the fictional --approval CHG-1042, the script only models the approval decision and hypothetical deduplication keys. Never copy this training approval string into a real production change workflow.
Run locally, not through JoeLearns
These commands must be run by you in a practice folder using Python 3.10 or newer. The web lesson only displays source, and the browser workbench cannot register OS tasks or use a local service identity.
"""JoeLearns Lesson 06: OFFLINE ONLY. No network calls, subprocesses or writes."""
import argparse
import json
from datetime import datetime, timezone
ALLOWED_MODES = ("plan", "simulate-apply")
SAMPLE = [
{"host": "SW-01", "site": "HQ", "state": "review"},
{"host": "AP-02", "site": "HQ", "state": "healthy"},
{"host": "FW-03", "site": "HQ", "state": "review"},
]
# Fictional approval record. A real system must fetch independently trusted approval.
APPROVAL = {
"id": "CHG-1042", "site": "HQ", "operation": "review",
"starts": "2026-11-10T01:00:00+00:00",
"ends": "2026-11-10T03:00:00+00:00",
"approved": True,
}
FIXED_NOW = datetime(2026, 11, 10, 2, tzinfo=timezone.utc)
def validate_config(config):
if not isinstance(config, dict) or set(config) != {"site", "mode", "limit"}:
raise ValueError("invalid-config")
if not isinstance(config["site"], str) or not config["site"].strip():
raise ValueError("invalid-site")
if config["mode"] not in ALLOWED_MODES:
raise ValueError("invalid-mode")
if type(config["limit"]) is not int or not 1 <= config["limit"] <= 25:
raise ValueError("invalid-limit")
return {"site": config["site"].strip().upper(),
"mode": config["mode"], "limit": config["limit"]}
def build_plan(records, site):
if not isinstance(records, list):
raise ValueError("invalid-records")
hosts = set()
proposed = []
for row in records:
if not isinstance(row, dict) or set(row) != {"host", "site", "state"}:
raise ValueError("invalid-record")
host, row_site, state = row["host"], row["site"], row["state"]
if (not isinstance(host, str) or not host.strip()
or not isinstance(row_site, str) or not row_site.strip()
or state not in ("healthy", "review")):
raise ValueError("invalid-record")
normalized = host.strip().upper()
if normalized in hosts:
raise ValueError("duplicate-host")
hosts.add(normalized)
if row_site.strip().upper() == site and state == "review":
proposed.append({"host": normalized, "operation": "review"})
return sorted(proposed, key=lambda item: item["host"])
def valid_approval(approval, site, now):
if not isinstance(approval, dict) or approval.get("approved") is not True:
return False
if approval.get("id") != "CHG-1042" or approval.get("site") != site:
return False
if approval.get("operation") != "review":
return False
try:
start = datetime.fromisoformat(approval["starts"])
end = datetime.fromisoformat(approval["ends"])
return (now.tzinfo is not None and now.utcoffset() is not None
and start.utcoffset() is not None and end.utcoffset() is not None
and start <= now < end)
except (ValueError, KeyError, TypeError):
return False
def run_job(config, records, approval, now, previous, seen=None, active=False):
"""Return decisions only: simulate-apply NEVER executes effects."""
seen = set() if seen is None else set(seen)
try:
checked = validate_config(config)
if active:
return {"state": "blocked", "reason": "overlap",
"previous": previous, "exit_code": 2}
actions = build_plan(records, checked["site"])
if len(actions) > checked["limit"]:
raise ValueError("over-limit")
except ValueError as exc:
return {"state": "blocked", "reason": str(exc),
"previous": previous, "exit_code": 2}
if checked["mode"] == "plan":
return {"state": "planned", "actions": actions,
"previous": previous, "exit_code": 0}
if not valid_approval(approval, checked["site"], now):
return {"state": "blocked", "reason": "approval",
"previous": previous, "exit_code": 2}
keys = [approval["id"] + ":review:" + item["host"] for item in actions]
new_keys = [key for key in keys if key not in seen]
return {"state": "simulated", "actions": actions,
"new_keys": new_keys, "duplicate_count": len(keys) - len(new_keys),
"previous": previous, "exit_code": 0}
def main(argv=None):
parser = argparse.ArgumentParser(description="Fictional offline scheduled-job simulator")
parser.add_argument("--mode", choices=ALLOWED_MODES, default="plan")
parser.add_argument("--scenario",
choices=("clean", "duplicate", "invalid", "overlap", "expired", "replay"),
default="clean")
parser.add_argument("--approval", default="")
args = parser.parse_args(argv)
records = [dict(item) for item in SAMPLE]
config = {"site": "HQ", "mode": args.mode, "limit": 5}
active = args.scenario == "overlap"
when = FIXED_NOW
seen = set()
if args.scenario == "duplicate":
records.append({"host": "sw-01", "site": "HQ", "state": "review"})
if args.scenario == "invalid":
config["limit"] = True
if args.scenario == "expired":
when = datetime(2026, 11, 10, 4, tzinfo=timezone.utc)
if args.scenario == "replay":
seen.add("CHG-1042:review:SW-01")
approved = APPROVAL if args.approval == "CHG-1042" else None
outcome = run_job(config, records, approved, when,
previous={"report": "last-known-good"}, seen=seen, active=active)
print(json.dumps(outcome, sort_keys=True, indent=2))
return outcome["exit_code"]
if __name__ == "__main__":
raise SystemExit(main())
python scheduled_job.pypython scheduled_job.py --mode simulate-apply --approval CHG-1042python scheduled_job.py --mode simulate-apply --scenario expired --approval CHG-1042python scheduled_job.py --scenario duplicatepython scheduled_job.py --mode simulate-apply --scenario replay --approval CHG-1042Verify that blocked runs exit with code 2 and that the previous fictional report is preserved. A planned or simulated outcome exits 0, but simulated never means actual work happened. The times and approval are intentionally fixed so you can reproduce the same decisions on any day. A real scheduler must use an accurately configured clock and an independently trusted approval system.
Now save the following file alongside it as test_scheduled_job.py and use unittest to check boundaries, replay, invalid input and overlap decisions. Change one expected result intentionally to verify the test really detects a failure, then restore it.
"""Run with python -m unittest test_scheduled_job.py -v (standard library)."""
import copy
import unittest
from datetime import datetime, timezone
from scheduled_job import APPROVAL, FIXED_NOW, SAMPLE, run_job
CONFIG = {"site": "HQ", "mode": "simulate-apply", "limit": 5}
PREVIOUS = {"report": "preserve-this"}
class ScheduledJobTests(unittest.TestCase):
def test_plan_never_requires_approval(self):
config = {**CONFIG, "mode": "plan"}
out = run_job(config, SAMPLE, None, FIXED_NOW, PREVIOUS)
self.assertEqual(out["state"], "planned")
self.assertEqual(len(out["actions"]), 2)
def test_simulated_apply_requires_approval(self):
out = run_job(CONFIG, SAMPLE, None, FIXED_NOW, PREVIOUS)
self.assertEqual((out["state"], out["reason"]), ("blocked", "approval"))
def test_valid_approval_simulates_without_writes(self):
original = copy.deepcopy(SAMPLE)
out = run_job(CONFIG, SAMPLE, APPROVAL, FIXED_NOW, PREVIOUS)
self.assertEqual(out["state"], "simulated")
self.assertEqual(len(out["new_keys"]), 2)
self.assertEqual(SAMPLE, original)
self.assertIs(out["previous"], PREVIOUS)
def test_wrong_site_is_not_authorized(self):
approval = {**APPROVAL, "site": "REMOTE"}
out = run_job(CONFIG, SAMPLE, approval, FIXED_NOW, PREVIOUS)
self.assertEqual(out["reason"], "approval")
def test_expired_change_window_blocks(self):
later = datetime(2026, 11, 10, 3, tzinfo=timezone.utc)
out = run_job(CONFIG, SAMPLE, APPROVAL, later, PREVIOUS)
self.assertEqual(out["reason"], "approval")
def test_duplicate_hosts_are_invalid(self):
records = SAMPLE + [{"host": "sw-01", "site": "HQ", "state": "review"}]
out = run_job(CONFIG, records, APPROVAL, FIXED_NOW, PREVIOUS)
self.assertEqual(out["reason"], "duplicate-host")
self.assertIs(out["previous"], PREVIOUS)
def test_replay_has_stable_action_keys(self):
seen = {"CHG-1042:review:SW-01"}
out = run_job(CONFIG, SAMPLE, APPROVAL, FIXED_NOW, PREVIOUS, seen)
self.assertEqual(out["duplicate_count"], 1)
self.assertEqual(out["new_keys"], ["CHG-1042:review:FW-03"])
def test_overlap_and_invalid_config_fail_closed(self):
overlap = run_job(CONFIG, SAMPLE, APPROVAL, FIXED_NOW, PREVIOUS, active=True)
malformed = run_job({**CONFIG, "limit": True}, SAMPLE, APPROVAL, FIXED_NOW, PREVIOUS)
self.assertEqual(overlap["reason"], "overlap")
self.assertEqual(malformed["reason"], "invalid-limit")
self.assertEqual(overlap["exit_code"], 2)
self.assertIs(malformed["previous"], PREVIOUS)
if __name__ == "__main__":
unittest.main()
python -m unittest test_scheduled_job.py -vpython -m unittest discover -vOptional real scheduler practice: After the tests pass, configure Windows Task Scheduler or a user-level cron/systemd timer in your own disposable environment to call python scheduled_job.py --mode plan. Record the interpreter, working directory, invocation identity, time zone, stdout/stderr and exit status. Do not configure a production host, credentials, live API endpoint, or a real apply action.
Optional local walkthrough: Create, verify and remove a read-only scheduled task (Windows, cron or systemd)
Complete the offline unit tests above first. Everything below is operating-system configuration, not Python code or something JoeLearns executes. Use a disposable practice account and project. Never schedule against customer records, production infrastructure or actual change-apply modes.
Windows · Task Scheduler
Run a read-only plan under your own account. The example assumes a disposable folder C:\Training\northstar-python-tool; replace it with your actual full path and the project's .venv interpreter.
- Manually verify the command in PowerShell first. Do not register a task until the local scheduled_job.py and unit tests run successfully.
- Open Task Scheduler → Create Task. Give it an unmistakable name such as Northstar Python PLAN (Training); select your own minimally privileged account.
- Under Triggers, add a short practice trigger and choose a recurrence appropriate to your experiment. Avoid the production environment.
- Under Actions → Start a program, set Program/script to the project .venv python.exe, Add arguments to the script's absolute path plus --mode plan, and Start in to the project directory (without quotation marks in that field).
- Save, run the task manually, inspect History and Last Run Result, then disable/delete this training task when finished.
Program/script: C:\Training\northstar-python-tool\.venv\Scripts\python.exe
Add arguments: "C:\Training\northstar-python-tool\scheduled_job.py" --mode plan
Start in: C:\Training\northstar-python-toolPowerShell or CMD: schtasks /Query /TN "Northstar Python PLAN (Training)" /V /FO LISTTask Scheduler → Task Scheduler Library → select the training task → Disable or Delete. Avoid modifying existing organization tasks.Linux / macOS · user crontab
A cron expression is NOT Python. The five fields are minute, hour, day of month, month and day of week, followed by an absolute command. Your shell's virtual-environment activation typically won't run automatically.
- Use a disposable local path with an existing scheduled_job.py. In the example, replace /home/learner with your own actual home directory and choose the interpreter in your own .venv.
- Manually run the planned command once and confirm it exits 0 without modifying any output files.
- Use crontab -e to edit only your own crontab. Add the training line below as a distinct entry. */30 in the minute field means minute 0 and 30 of each hour.
- Use crontab -l to verify your actual entry. Observe execution and output through the local mechanism your cron service provides; some environments send output through local mail rather than a terminal.
- Remove ONLY the Northstar training line using crontab -e when you're finished. Do not run crontab -r (which removes every entry).
*/30 * * * * /home/learner/northstar-python-tool/.venv/bin/python /home/learner/northstar-python-tool/scheduled_job.py --mode plancrontab -lcrontab -e → remove only the Northstar training line → save → crontab -l to verify removal. Check whether cron is actually installed and running on your system.Linux · systemd user service and timer
A .service defines what to launch; a .timer decides when. This is for a user-level unit on a Linux machine with an active systemd user manager, not an organization service or a production deployment.
- First verify the plan command manually. In your disposable project, use a .venv and scheduled_job.py from the lesson.
- Create ~/.config/systemd/user/northstar-plan.service and northstar-plan.timer with the contents below. The %h specifier represents the user home directory; adjust the folder name to your actual project.
- Reload user units, then enable and start the timer—not the service itself. The hourly timer triggers the oneshot service under your current user.
- Inspect list-timers, status and journal entries. The service will not automatically remain active after its oneshot run; that is expected.
- Disable/stop the training timer and remove only those two Northstar files when finished; reload the user daemon. Do not enable lingering, elevate privileges or install a system-wide service for this lesson.
# ~/.config/systemd/user/northstar-plan.service
[Unit]
Description=Read-only Northstar Python training plan
[Service]
Type=oneshot
WorkingDirectory=%h/northstar-python-tool
ExecStart=%h/northstar-python-tool/.venv/bin/python %h/northstar-python-tool/scheduled_job.py --mode plan
# ~/.config/systemd/user/northstar-plan.timer
[Unit]
Description=Hourly Northstar Python training plan
[Timer]
OnCalendar=hourly
Persistent=false
[Install]
WantedBy=timers.targetsystemctl --user daemon-reload
systemctl --user enable --now northstar-plan.timer
systemctl --user list-timers --all
journalctl --user -u northstar-plan.service -n 20 --no-pagersystemctl --user disable --now northstar-plan.timer
# Then delete only your northstar-plan.service and northstar-plan.timer files.
systemctl --user daemon-reloadBefore considering a real scheduled job: test missing working directories, mismatched interpreters, nonzero exit codes, time-zone/DST effects and an overlapping run. A successful scheduler trigger is never proof that a write or device change is authorized.
Practice controlled automation in real Python
These eight original coding challenges move from schedule validation to UTC maintenance windows, config precedence, stable plans, scoped approvals, idempotent keys, overlap decisions and a final integrated controlled job. Run your own code, investigate errors and Check against changed input cases in the isolated Python workbench.
Write it. Run it. Understand it.
Real Python execution, a variable trace and feedback against the requirements. You write the code—answers aren't revealed up front.
Validate a bounded schedule interval
Write interval_minutes(value): accept only an integer between 5 and 1440 inclusive, returning the value, or return None for booleans, fractions, strings and out-of-range values. The scheduler calls this helper before registering a recurrence; do not silently cast a user-supplied string. Print the interval of the sample job. The checker tests minimum, maximum, malformed types and unexpected values.
15 minutesTry 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.
Make the safe decision without relying on a happy-path example.
Try expired approval records, invalid limits, repeated action keys and overlapping scheduled runs. Your simulated implementation must preserve old verified output whenever a safety gate blocks the job.
Continue to all eight scheduling challenges →Continue to the final Northstar Fleet Audit →Return to the Applied Python path →Official Python references
These standard-library references explain the data types, configuration, log and testing behavior demonstrated above. Scheduler documentation differs by operating system; consult the platform's own documentation before deploying a real task.