Compose 项目把资源声明与外部输入组合起来

from pathlib import Path
from tempfile import mkdtemp
from urllib.request import urlopen
import json,os,subprocess,sys,time,uuid

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

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-project-'+uuid.uuid4().hex;environment=os.environ.copy();environment.pop('TASK_COMPOSE_GREETING',None)
root=None;project_started=False
label='com.docker.compose.project='+name
kinds=[('container',['ps','-aq'],2),('network',['network','ls','-q'],1),('volume',['volume','ls','-q'],1)]
def listed(command):return docker(*command,'--filter','label='+label,timeout=10).stdout.split()
def clear_kind(kind,command,limit):
    identities=listed(command)
    if len(identities)>limit:raise RuntimeError('unexpected project resource count; inspect '+name)
    cleanup_steps([(kind+' '+identity,lambda identity=identity:docker(kind,'rm',*(['-f'] if kind=='container' else []),identity,timeout=10)) for identity in identities],None)
for kind,command,limit in kinds:require_empty(listed(command))
print('fixture project:',name,flush=True)
try:
    root=Path(mkdtemp(prefix=name+'-'));file=root/'compose.yaml'
    def compose(*args,check=True,env=None,timeout=120):return run(['docker','compose','-p',name,'--env-file',str(root/'.env'),'-f',str(file),*args],check,environment if env is None else env,timeout)
    def container(service):return compose('ps','-q',service).stdout.strip()
    def inspect(identity):return json.loads(docker('inspect',identity).stdout)[0]
    def request():
        port=inspect(container('api'))['NetworkSettings']['Ports']['8080/tcp'][0]
        assert port['HostIp']=='127.0.0.1'
        deadline=time.monotonic()+10
        while time.monotonic()<deadline:
            try:
                with urlopen('http://127.0.0.1:'+port['HostPort'],timeout=1) as response:return response.read().decode()
            except OSError:time.sleep(.1)
        raise AssertionError('published endpoint unavailable')
    server='from http.server import BaseHTTPRequestHandler,HTTPServer\nimport os\nclass H(BaseHTTPRequestHandler):\n def do_GET(self):self.send_response(200);self.end_headers();self.wfile.write(os.environ["GREETING"].encode())\n def log_message(self,*args):pass\nHTTPServer(("0.0.0.0",8080),H).serve_forever()\n'
    spec={'services':{'api':{'image':BASE,'command':['python','-c',server],'environment':{'GREETING':'${TASK_COMPOSE_GREETING:-default}'},'ports':[{'target':8080,'published':'0','host_ip':'127.0.0.1'}],'volumes':['state:/state']},'client':{'image':BASE,'command':['python','-c','import time;time.sleep(180)']}},'volumes':{'state':{}}}
    file.write_text(json.dumps(spec,indent=2));(root/'.env').write_text('TASK_COMPOSE_GREETING=from-file\n')
    rendered=json.loads(compose('config','--format','json').stdout);assert rendered['services']['api']['environment']['GREETING']=='from-file'
    override=environment|{'TASK_COMPOSE_GREETING':'from-shell'}
    rendered=json.loads(compose('config','--format','json',env=override).stdout);assert rendered['services']['api']['environment']['GREETING']=='from-shell'
    project_started=True
    compose('up','-d',env=override);assert request()=='from-shell'
    api=container('api');client=container('client');first_network=next(iter(inspect(api)['NetworkSettings']['Networks'].values()))['NetworkID']
    result=docker('exec',client,'python','-c','import urllib.request;print(urllib.request.urlopen("http://api:8080",timeout=2).read().decode())')
    assert result.stdout.strip()=='from-shell'
    docker('exec',client,'python','-c','import socket;s=socket.socket();assert s.connect_ex(("127.0.0.1",8080))!=0;s.close()')
    docker('exec',api,'python','-c','from pathlib import Path;Path("/state/record").write_text("preserved")')
    volume=next(m['Name'] for m in inspect(api)['Mounts'] if m['Destination']=='/state')
    assert volume==name+'_state'
    compose('down');assert docker('network','inspect',first_network,check=False).returncode!=0
    assert docker('volume','inspect',volume,check=False).returncode==0
    compose('up','-d',env=override);assert request()=='from-shell'
    api=container('api');assert next(iter(inspect(api)['NetworkSettings']['Networks'].values()))['NetworkID']!=first_network
    docker('exec',api,'python','-c','from pathlib import Path;assert Path("/state/record").read_text()=="preserved"')
    assert docker('exec',container('client'),'python','-c','import urllib.request;print(urllib.request.urlopen("http://api:8080",timeout=2).read().decode())').stdout.strip()=='from-shell'
    print('shell interpolation overrides .env; internal clients resolve api:8080 while localhost remains their own container')
    print('the host endpoint is published only on loopback; down/up recreates the project network and retains the named data volume')
finally:
    primary=sys.exc_info()[1];steps=[]
    if project_started:
        steps.append(('compose down',lambda:compose('down','--volumes','--remove-orphans','--timeout','5',timeout=10)))
        steps.extend((kind+' removal',lambda kind=kind,command=command,limit=limit:clear_kind(kind,command,limit)) for kind,command,limit in kinds)
        steps.extend((kind+' absence',lambda command=command:require_empty(listed(command))) for kind,command,limit in kinds)
    if root is not None:
        steps.extend([('env file',lambda:remove_file(root/'.env')),('compose file',lambda:remove_file(root/'compose.yaml')),('temporary directory',root.rmdir)])
    cleanup_steps(steps,primary)