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

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

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

name='xbcatm-project-'+uuid.uuid4().hex[:10];environment=os.environ.copy();environment.pop('TASK_COMPOSE_GREETING',None)
with TemporaryDirectory(prefix=name+'-') as directory:
    root=Path(directory);file=root/'compose.yaml'
    def compose(*args,check=True,env=None):return run(['docker','compose','-p',name,'--env-file',str(root/'.env'),'-f',str(file),*args],check,env or environment)
    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')
    try:
        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'
        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:compose('down','--volumes','--remove-orphans',check=False)