proxy_pass 既选择上游，也映射请求 URI

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_text(content)
            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()
            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'

backend=r'''from http.server import BaseHTTPRequestHandler,HTTPServer
import json
class Handler(BaseHTTPRequestHandler):
 def do_GET(self):
  body=self.rfile.read(int(self.headers.get('Content-Length','0')))
  document={'path':self.path,'method':self.command,'body':body.decode(),'headers':{key:self.headers.get(key) for key in ['Host','X-Forwarded-For','X-Real-IP','X-Forwarded-Proto','X-Forwarded-Host']}}
  data=json.dumps(document).encode()
  self.send_response(200);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(data)));self.send_header('X-Backend-Info','fixture-node');self.end_headers();self.wfile.write(data)
 do_POST=do_GET
 def log_message(self,*args):pass
HTTPServer(('0.0.0.0',8081),Handler).serve_forever()
'''

locations=r'''
location /keep/ { proxy_pass http://backend:8081; }
location /strip/ { proxy_pass http://backend:8081/; }
location /replace/ { proxy_pass http://backend:8081/v1/; }
location /join/ { proxy_pass http://backend:8081/v1; }
location /variable/ { set $destination "http://fixture_pool/fixed?chosen=1"; proxy_pass $destination; }
'''
invalid=config(r'location ~ ^/bad/ { proxy_pass http://backend:8081/replacement/; }')
with lab(config(locations,'upstream fixture_pool { server backend:8081; }'),{'invalid.conf':invalid},backend) as (container,request):
    cases={'/keep/item?x=1':'/keep/item?x=1','/strip/item?x=1':'/item?x=1','/replace/item?x=1':'/v1/item?x=1','/join/item?x=1':'/v1item?x=1','/keep/a%2Fb?x=1':'/keep/a%2Fb?x=1','/strip/a%2Fb?x=1':'/a/b?x=1','/variable/item?original=1':'/fixed?chosen=1'}
    for path,expected in cases.items():
        status,headers,body=request(path);assert status==200 and json.loads(body)['path']==expected,(path,status,body)
    invalid_result=docker('exec',container,'nginx','-t','-c','/srv/invalid.conf',check=False)
    assert invalid_result.returncode!=0 and 'cannot have URI part in location given by regular expression' in invalid_result.stderr
    print('no URI preserves the original path; a literal URI replaces the normalized matching prefix, including its slash boundaries')
    print('a variable-supplied URI is used as specified; regex-location URI replacement is rejected by nginx -t')