显式启用 gzip，再验证协商与解码后的字节

from contextlib import contextmanager
from pathlib import Path
from tempfile import TemporaryDirectory
import http.client,json,socket,subprocess,time,uuid

NGINX='nginx@sha256:552e7481ca93ffccd046aa658dbbed22caefbc09c66fa7cd247cbb90b8a5c609'
PYTHON='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84'
def docker(*args,check=True):
    result=subprocess.run(['docker',*args],capture_output=True,text=True,timeout=120)
    if check:assert result.returncode==0,result.stderr
    return result
def wait(predicate):
    deadline=time.monotonic()+15
    while time.monotonic()<deadline:
        try:
            if predicate():return
        except OSError:pass
        time.sleep(.1)
    raise AssertionError('fixture did not become ready')

@contextmanager
def lab(configuration,files=None,backend=None):
    name='xbcatm-nginx-'+uuid.uuid4().hex[:10];tags=[name+':proxy',name+':backend'];containers=[name+'-proxy',name+'-backend']
    try:
        with TemporaryDirectory(prefix=name+'-') as directory:
            root=Path(directory);(root/'files').mkdir();(root/'nginx.conf').write_text(configuration)
            for path,content in (files or {}).items():
                target=root/'files'/path;target.parent.mkdir(parents=True,exist_ok=True)
                target.write_bytes(content if isinstance(content,bytes) else content.encode())
            docker('network','create',name)
            if backend is not None:
                (root/'backend.py').write_text(backend)
                (root/'backend.Dockerfile').write_text(f'FROM {PYTHON}\nCOPY backend.py /backend.py\nCMD ["python", "/backend.py"]\n')
                docker('build','--network','none','-f',str(root/'backend.Dockerfile'),'-t',tags[1],str(root))
                docker('run','-d','--name',containers[1],'--network',name,'--network-alias','backend',tags[1])
                wait(lambda:docker('exec',containers[1],'python','-c','import socket;s=socket.create_connection(("127.0.0.1",8081),timeout=.5);s.close()',check=False).returncode==0)
            (root/'Dockerfile').write_text(f'FROM {NGINX}\nCOPY nginx.conf /etc/nginx/nginx.conf\nCOPY files /srv\n')
            docker('build','--network','none','-t',tags[0],str(root))
            docker('run','--rm','--network',name,tags[0],'nginx','-t')
            docker('run','-d','--name',containers[0],'--network',name,'-p','127.0.0.1::8080',tags[0])
            details=json.loads(docker('inspect',containers[0]).stdout)[0]
            binding=details['NetworkSettings']['Ports']['8080/tcp'][0];assert binding['HostIp']=='127.0.0.1';port=int(binding['HostPort'])
            def ready():
                connection=socket.create_connection(('127.0.0.1',port),timeout=.5);connection.close();return True
            wait(ready)
            def request(path,method='GET',headers=None,body=None):
                connection=http.client.HTTPConnection('127.0.0.1',port,timeout=5)
                try:
                    connection.request(method,path,body=body,headers=headers or {})
                    response=connection.getresponse();return response.status,{k.lower():v for k,v in response.getheaders()},response.read()
                finally:connection.close()
            request.port=port
            yield containers[0],request
    finally:
        docker('rm','-f',*containers,check=False);docker('image','rm',*tags,check=False);docker('network','rm',name,check=False)

def config(locations,extra=''):
    return 'worker_processes 1;\nerror_log /tmp/error.log notice;\nevents {}\nhttp {\naccess_log off;\n'+extra+'\nserver { listen 8080; root /srv;\n'+locations+'\n}\n}\n'
import gzip,hashlib
backend=r'''
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
class Handler(BaseHTTPRequestHandler):
    protocol_version='HTTP/1.1'
    def log_message(self,*args):pass
    def do_GET(self):
        self.send_response(200);self.send_header('Content-Type','application/json')
        self.send_header('Connection','close');self.end_headers();self.wfile.write(b'{"ok":true}\n');self.close_connection=True
ThreadingHTTPServer(('0.0.0.0',8081),Handler).serve_forever()
'''
payload=json.dumps({'message':'compressible nginx response '*1000}).encode()
locations='''
location = /disabled.json { gzip off; gzip_types application/json; }
location = /novary.json { gzip_vary off; }
location = /stream { proxy_pass http://backend:8081; }
'''
extra='''
types { application/json json; text/html html; application/octet-stream bin; }
gzip on; gzip_types application/json; gzip_vary on; gzip_min_length 1024; gzip_comp_level 3;
'''
files={name:payload for name in ['large.json','other.bin','page.html','disabled.json','novary.json']};files['tiny.json']=b'{"ok":true}\n'
with lab(config(locations,extra),files,backend) as (_,request):
    results={}
    for encoding in ['gzip','identity','gzip;q=0','br']:
        status,h,b=request('/large.json',headers={'Accept-Encoding':encoding});assert status==200
        if encoding=='gzip':assert h['content-encoding']=='gzip' and gzip.decompress(b)==payload and len(b)<len(payload)
        else:assert 'content-encoding' not in h and b==payload
        assert h['vary']=='Accept-Encoding'
        results[encoding]={'encoding':h.get('content-encoding','identity'),'body_bytes':len(b)}
    for path,encoded in [('/tiny.json',False),('/other.bin',False),('/page.html',True),('/disabled.json',False),('/novary.json',True),('/stream',True)]:
        status,h,b=request(path,headers={'Accept-Encoding':'gzip'});assert status==200
        assert (h.get('content-encoding')=='gzip')==encoded,(path,h)
        decoded=gzip.decompress(b) if encoded else b
        assert decoded==(files['tiny.json'] if path=='/stream' else files[path[1:]])
        if path=='/novary.json':assert 'vary' not in h
        results[path]={'encoding':h.get('content-encoding','identity'),'body_bytes':len(b),'vary':h.get('vary')}
    results['decoded_sha256']=hashlib.sha256(payload).hexdigest()
    print(json.dumps(results,sort_keys=True))