ENTRYPOINT 和 CMD 定义可被运行者覆盖的启动默认值

from pathlib import Path
from tempfile import mkdtemp
import json,subprocess,sys,time,uuid

BASE='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84'
def docker(*args,check=True,timeout=120):
    result=subprocess.run(['docker',*args],capture_output=True,text=True,timeout=timeout)
    if check:assert result.returncode==0,result.stderr
    return result
def inspect(name):return json.loads(docker('inspect',name).stdout)[0]

def cleanup_steps(steps, primary):
    failures=[]
    for label, action in steps:
        try:action()
        except BaseException as error:
            failures.append(label+': '+type(error).__name__+': '+str(error))
    if failures:
        message='cleanup incomplete or uncertain: '+'; '.join(failures)
        try:print(message,file=sys.stderr)
        except (OSError,ValueError):pass
        if primary is None:raise RuntimeError(message)
    return failures

def remove_file(path):
    try:path.unlink()
    except FileNotFoundError:pass

def require_empty(value):
    if value:raise RuntimeError('residual resources: '+str(value))

name='xbcatm-entrypoint-'+uuid.uuid4().hex;tag=name+':probe'
containers=[name+'-exec',name+'-child'];probes=[name+'-args-'+str(i) for i in range(4)]
all_names=probes+containers;attempted=[];build_started=False;root=None
def current_names():return docker('ps','-a','--format','{{.Names}}',timeout=10).stdout.splitlines()
def remaining_names():return sorted(set(all_names)&set(current_names()))
require_empty(remaining_names());require_empty(docker('image','ls','-q',tag,timeout=10).stdout.strip())
def launch(container,*args):
    attempted.append(container)
    return docker('run','--name',container,*args)
print('fixture containers:',all_names,'image tag:',tag,flush=True)
try:
    root=Path(mkdtemp(prefix=name+'-'))
    (root/'args.py').write_text('import json,sys\nprint(json.dumps(sys.argv[1:]))\n')
    (root/'service.py').write_text('import json,os,signal\ndef stop(signum,frame):\n print("stopped",flush=True)\n raise SystemExit(0)\nsignal.signal(signal.SIGTERM,stop)\nprint(json.dumps({"pid":os.getpid()}),flush=True)\nwhile True:signal.pause()\n')
    (root/'exec.sh').write_text('#!/bin/sh\nexec python /service.py "$@"\n')
    (root/'child.sh').write_text('#!/bin/sh\npython /service.py "$@" &\nwait $!\n')
    (root/'Dockerfile').write_text(f'FROM {BASE}\nCOPY args.py service.py exec.sh child.sh /\nRUN chmod +x /exec.sh /child.sh\nENV DEMO=expanded\nENTRYPOINT ["python", "/args.py"]\nCMD ["default", "$DEMO"]\n')
    build_started=True
    docker('build','--network','none','-t',tag,str(root))
    assert json.loads(launch(probes[0],'--network','none',tag).stdout)==['default','$DEMO']
    assert json.loads(launch(probes[1],'--network','none',tag,'changed').stdout)==['changed']
    assert launch(probes[2],'--network','none','--entrypoint','python',tag,'-c','print("override")').stdout.strip()=='override'
    assert launch(probes[3],'--network','none','--entrypoint','/bin/sh',tag,'-c','printf "%s" "$DEMO"').stdout=='expanded'
    print('CMD supplies literal default arguments; trailing arguments replace it; --entrypoint replaces the executable')
    for container,wrapper in zip(containers,['/exec.sh','/child.sh']):
        launch(container,'-d','--network','none','--entrypoint',wrapper,tag)
        deadline=time.monotonic()+10
        while time.monotonic()<deadline:
            logs=docker('logs',container).stdout
            if logs.strip():break
            time.sleep(.05)
        else:raise AssertionError('service did not become ready')
        pid=json.loads(logs.splitlines()[0])['pid']
        assert (pid==1)==(wrapper=='/exec.sh')
        docker('stop','--time','1',container)
        state=inspect(container)['State'];logs=docker('logs',container).stdout
        if wrapper=='/exec.sh':assert state['ExitCode']==0 and 'stopped' in logs
        else:assert state['ExitCode']==137 and 'stopped' not in logs
    print('exec makes the handler PID 1 and stops cleanly; the non-forwarding shell wrapper leaves its child unnotified until forced termination')
finally:
    primary=sys.exc_info()[1]
    steps=[('container '+container,lambda container=container:docker('rm','-f',container,timeout=10)) for container in attempted]
    if build_started:steps.append(('image tag',lambda:docker('image','rm','--no-prune',tag,timeout=10)))
    steps.append(('container absence',lambda:require_empty(remaining_names())))
    if build_started:steps.append(('image-tag absence',lambda:require_empty(docker('image','ls','-q',tag,timeout=10).stdout.strip())))
    if root is not None:
        steps.extend((filename,lambda filename=filename:remove_file(root/filename)) for filename in ['args.py','service.py','exec.sh','child.sh','Dockerfile'])
        steps.append(('temporary directory',root.rmdir))
    cleanup_steps(steps,primary)