LEARN. PRACTICE. GROW.Lab contract
An APS team prepares two reconciliation workers. The change request requires evidence of target selection and modified files. Before rehearsal, write three criteria: each worker has its own identity in the file; an unchanged repeat does not activate configuration again; an invalid proposal preserves previous state. These criteria make concrete effects observable. Exit code zero without inspecting the destination does not answer all three. This course lab uses two inventory names connected through connection=local to one machine. Each name has a separate directory. This arrangement studies per-host execution but does not prove machine isolation, SSH access or resilience.
Prepare and execute
The content/labs/rhce-execution directory contains playbooks, template, validator and run.py. Create an isolated Python environment and install ansible-core 2.16.14 to reproduce the recorded execution version. Then invoke run.py with --ansible-bin pointing to ansible-playbook and --output naming the intended report. The runner creates a temporary inventory, pins the interpreter and supplies lab_root through extra vars. The selection rehearsal uses --list-hosts with --limit worker_a and confirms worker_b is absent. Only then does it execute both targets. The code below is the runner that generates this inventory; supply the actual ansible-playbook path in your environment.
Identity and observation
Compare the bytes of both worker.conf files: their worker fields should differ. Compare activation logs too, not just the recap. If both wrote to one path, two per-host results could conceal replacement of the same file. In a real project, also record the mapping between alias, address and role. A task delegated to the controller may still be invoked by several hosts in parallel. For a shared record, choose one aggregation or explicit serialization suited to the requirement; changing delegate_to alone does not prevent concurrency. The lab does not force a random race and claim it has been fixed.
Acceptance and transfer to RHEL
The report identifies the Ansible version, operating system, file hashes and executed checks. Do not use it as reboot or systemd service evidence: neither behavior is exercised here. For the next phase, apply the same contract to disposable RHEL machines with operational access identities. Add checks of service behavior, persistence and actual permissions. Give the RUN team the command, version and expected results so it can repeat the rehearsal without the author’s laptop or personal account.
# Save as run.py. Requires the other five lab files in the same directory.
# Usage: python3 run.py --ansible-bin /absolute/venv/bin/ansible-playbook --output evidence.json
"""Original local-only Ansible lab. No SSH, become, services or real data."""
import argparse,hashlib,json,os,pathlib,platform,re,subprocess,tempfile,zipfile
from datetime import datetime,timezone
p=argparse.ArgumentParser;p.add_argument('--ansible-bin',required=True);p.add_argument('--output',required=True);args=p.parse_args
source=pathlib.Path(__file__).resolve.parent;binary=pathlib.Path(args.ansible_bin).resolve;checks=[];runs=[]
def require(name,condition):
assert condition,name
checks.append({'name':name,'passed':True})
with tempfile.TemporaryDirectory(prefix='dr-rhce-lab-') as temporary:
root=pathlib.Path(temporary);(root/'home').mkdir;inventory=root/'inventory.ini'inventory.write_text('[workers]\nworker_a ansible_connection=local\nworker_b ansible_connection=local\n[workers:vars]\nansible_python_interpreter='+str(binary.parent/'python')+'\n')
env={**os.environ,'ANSIBLE_HOME':str(root/'home'),'ANSIBLE_LOCAL_TEMP':str(root/'local'),'ANSIBLE_REMOTE_TEMP':str(root/'remote'),'ANSIBLE_NOCOLOR':'1','ANSIBLE_CONFIG':str(root/'ansible.cfg'),'OBJC_DISABLE_INITIALIZE_FORK_SAFETY':'YES'}
(root/'ansible.cfg').write_text('[defaults]\nretry_files_enabled=False\nhost_key_checking=True\n')
version=subprocess.check_output([str(binary),'--version'],env=env,text=True).splitlines[0]
def run(label,play,extra=None,flags=,expected=0):
variables={'lab_root':str(root/'nodes'),'service_port':8443,'use_marker':False,**(extra or {})}
command=[str(binary),'-i',str(inventory),str(source/play),'-e',json.dumps(variables),*flags]
r=subprocess.run(command,env=env,text=True,capture_output=True,timeout=60)
assert r.returncode==expected,(label,r.returncode,r.stdout,r.stderr)
recaps={host:{k:int(v) for k,v in re.findall(r'(ok|changed|unreachable|failed|skipped|rescued|ignored)=(\d+)',line)} for line in r.stdout.splitlines for host in ['worker_a','worker_b'] if line.startswith(host+' ') and 'ok=' in line}
runs.append({'label':label,'returnCode':r.returncode,'recaps':recaps,'stdoutSha256':hashlib.sha256(r.stdout.encode).hexdigest});return r,recaps
run('syntax','converge.yml',flags=['--syntax-check'])
listed,_=run('limited-selection','converge.yml',flags=['--list-hosts','--limit','worker_a'])
require('limit resolves only worker_a','worker_a' in listed.stdout and 'worker_b' not in listed.stdout)
run('first-convergence','converge.yml')
def contents(name):return {h:(root/'nodes'/h/name).read_bytes for h in ['worker_a','worker_b']}
initial=contents('worker.conf');activation=contents('activation.log')
require('separate identities have distinct configurations',initial['worker_a']!=initial['worker_b'])
require('first activation once per worker',all(x==b'activated\n' for x in activation.values))
_,recap=run('repeat-convergence','converge.yml');require('second convergence reports no change',all(x['changed']==0 for x in recap.values) and len(recap)==2)
require('repeat leaves configuration and activation unchanged',contents('worker.conf')==initial and contents('activation.log')==activation)
run('invalid-template','converge.yml',{'service_port':70000},expected=2)
require('invalid candidate preserves destinations',contents('worker.conf')==initial)
require('invalid candidate never activates',contents('activation.log')==activation)
run('check-prediction','converge.yml',{'service_port':9443},flags=['--check'])
require('check leaves actual state unchanged',contents('worker.conf')==initial and contents('activation.log')==activation)
for h in ['worker_a','worker_b']:
release=root/'nodes'/h/'release'release.mkdir;(release/'obsolete.txt').write_text('old helper\n')
bundle=root/'release.zip'
def pack(text):
with zipfile.ZipFile(bundle,'w') as z:z.writestr('payload.txt',text)
pack('release-one\n');run('archive-first','archive.yml',{'bundle_path':str(bundle)})
require('archive contains first payload',all(x==b'release-one\n' for x in contents('release/payload.txt').values))
require('archive does not prune unlisted files',all(x==b'old helper\n' for x in contents('release/obsolete.txt').values))
_,recap=run('archive-repeat','archive.yml',{'bundle_path':str(bundle)})
require('unchanged zip repeat reports no change',len(recap)==2 and all(x['changed']==0 for x in recap.values))
for h in ['worker_a','worker_b']:(root/'nodes'/h/'release/ready').write_text('old marker\n')
pack('release-two\n');run('stale-marker','archive.yml',{'bundle_path':str(bundle),'use_marker':True})
require('existing marker suppresses new archive',all(x==b'release-one\n' for x in contents('release/payload.txt').values))
run('archive-without-marker','archive.yml',{'bundle_path':str(bundle)})
require('new archive changes intended payload',all(x==b'release-two\n' for x in contents('release/payload.txt').values))
require('new archive still preserves extra file',all(x==b'old helper\n' for x in contents('release/obsolete.txt').values))
run('failed-release-recovery','recovery.yml',expected=2)
require('rescue restores exact previous bytes',contents('worker.conf')==initial)
require('always evidence written after rescue failure',all(x==b'recovery-path-visited\n' for x in contents('recovery.log').values))
require('recovery did not claim service activation',contents('activation.log')==activation)
report={'checkedAt':datetime.now(timezone.utc).isoformat,'ansible':version,'platform':platform.platform,'localAliases':2,'actualMachines':1,'runs':runs,'checks':checks,'sourceHashes':{f.name:hashlib.sha256(f.read_bytes).hexdigest for f in sorted(source.iterdir) if f.suffix in ['.yml','.j2','.py']},'limitations':['Local filesystem execution on macOS; no RHEL managed node or AAP execution environment.','Activation log is a synthetic handler effect, not a service restart or health check.','No SSH, privilege escalation, reboot, storage devices or SELinux validation.'],'cleanup':'Owned temporary target tree removed by TemporaryDirectory.'}
pathlib.Path(args.output).write_text(json.dumps(report,indent=2)+'\n');print(json.dumps({'runs':len(runs),'checks':len(checks),'ansible':version}))
Two local aliases produce different files; this demonstrates selection and path separation, not two machines.
Common pitfalls
Confusing inventory host with physical machine; using recap as sole evidence; delegating shared writes without controlling concurrency.
Related topics: Validated configuration and observable activation · Archives, residual files and markers · Recovery and change outcome
State the contract, measure effects and limit the conclusion to the exercised environment.
Reference: Delegation and local execution · EX294 current objectives inspected 2026-09-30; RHCE in Ansible framework effective 2026-05-11; booking product version not publicly pinned