redis-core/examples/run-redis-fixture.py

Version a84871d065f3.bb1 · MIT. This preview displays packaged text and does not execute code. Treat the contents as untrusted instructions.

← Return to resource and package checksum

"""Own disposable Docker resources only. MIT, Copyright 2026 BB Skills."""
from pathlib import Path
import subprocess,json,datetime as dt,hashlib,secrets,time
b=Path(__file__).resolve().parent;net='bb-v44-redis-net';server='bb-v44-redis';probe='bb-v44-redis-probe';created=[]
def run(args,**kw):return subprocess.run(args,capture_output=True,text=True,timeout=kw.pop('timeout',60),**kw)
def checked(args,**kw):
 r=run(args,**kw)
 if r.returncode:raise RuntimeError('Docker operation failed: '+r.stderr[-2000:])
 return r.stdout
for kind,name in [('network',net),('container',server),('container',probe)]:assert run(['docker',kind,'inspect',name]).returncode!=0,'Existing fixture name: '+name
images={}
for tag in ['redis:7-alpine','bb-skills:20261004-v41']:
 row=json.loads(checked(['docker','image','inspect',tag]))[0];images[tag]={'id':row['Id'],'digests':row['RepoDigests']}
try:
 checked(['docker','network','create','--internal',net]);created.append(('network',net))
 checked(['docker','run','-d','--name',server,'--network',net,'--memory','256m','--cpus','1','--tmpfs','/data:rw,size=134217728','--tmpfs','/tmp:rw,size=16777216','redis:7-alpine','redis-server','--save','','--appendonly','no','--protected-mode','yes','--bind','0.0.0.0']);created.append(('container',server))
 for _ in range(20):
  if run(['docker','exec',server,'redis-cli','PING']).stdout.strip()=='PONG':break
  time.sleep(0.5)
 else:raise RuntimeError('Temporary Redis not ready')
 bootstrap=secrets.token_hex(24)
 assert checked(['docker','exec','-i',server,'redis-cli','--raw'],input='ACL SETUSER default reset on >'+bootstrap+' ~* &* +@all\n').strip()=='OK'
 checked(['docker','run','-i','--name',probe,'--network',net,'--memory','128m','--cpus','1','--read-only','--cap-drop','ALL','--security-opt','no-new-privileges','--tmpfs','/tmp:rw,size=16777216','--mount','type=bind,source='+str(b)+',target=/fixture,readonly','--entrypoint','python','bb-skills:20261004-v41','/fixture/probe-redis.py'],input=json.dumps({'bootstrap_password':bootstrap})+'\n',timeout=90)
 created.append(('container',probe))
 raw=checked(['docker','logs',probe]);data=json.loads(raw);assert all(c['status']=='passed' for c in data['cases'])
 data['images']=images;data['fixture']={'internal_network':True,'published_ports':False,'production_mounts':False,'persistent_data':False,'synthetic_only':True,'network_bind_note':'Container-only 0.0.0.0 within a dedicated internal Docker network; not a production network configuration.'}
 data['source_hashes']['run-redis-fixture.py']=hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
 (b/'native-redis-evidence.json').write_text(json.dumps(data,indent=2),encoding='utf-8',newline='\n')
 print(json.dumps({'observations':data['observations'],'redis_version':data['redis_version'],'full_skill_runtime_tested':False,'all_recorded_cases_passed':True}))
finally:
 # The probe is owned by this attempt after docker run even if its exit code is nonzero.
 if any(v==('container',server) for v in created) and run(['docker','container','inspect',probe]).returncode==0 and ('container',probe) not in created:created.append(('container',probe))
 for kind,name in reversed(created):
  r=run(['docker','rm','-fv',name] if kind=='container' else ['docker','network','rm',name]);assert r.returncode==0,(kind,name)