10.1.x: Stabilize post and slow_post autests under parallel runs (#12886)

tests/gold_tests/post/post-early-return.test.py

@@ -61,18 +61,27 @@
         'proxy.config.diags.debug.tags': 'http',
     })
 
+mock_origin = os.path.join(Test.Variables.AtsTestToolsDir, 'mock_origin.py')
+mock_origin_args = '--status 420 --reason "Be Calm"'
+
 server1 = Test.Processes.Process(
-    "server1", "bash -c '" + Test.TestDirectory + "/server1.sh {} outserver1'".format(Test.Variables.upstream_port1))
+    "server1", f"python3 {mock_origin} {Test.Variables.upstream_port1} {mock_origin_args} --output outserver1")
 server2 = Test.Processes.Process(
-    "server2", "bash -c '" + Test.TestDirectory + "/server1.sh {} outserver1'".format(Test.Variables.upstream_port2))
+    "server2", f"python3 {mock_origin} {Test.Variables.upstream_port2} {mock_origin_args} --output outserver1")
 server3 = Test.Processes.Process(
-    "server3", "bash -c '" + Test.TestDirectory + "/server1.sh {} outserver1'".format(Test.Variables.upstream_port3))
+    "server3", f"python3 {mock_origin} {Test.Variables.upstream_port3} {mock_origin_args} --output outserver1")
 server4 = Test.Processes.Process(
-    "server4", "bash -c '" + Test.TestDirectory + "/server1.sh {} outserver1'".format(Test.Variables.upstream_port4))
+    "server4", f"python3 {mock_origin} {Test.Variables.upstream_port4} {mock_origin_args} --output outserver1")
 server5 = Test.Processes.Process(
-    "server5", "bash -c '" + Test.TestDirectory + "/server1.sh {} outserver1'".format(Test.Variables.upstream_port5))
+    "server5", f"python3 {mock_origin} {Test.Variables.upstream_port5} {mock_origin_args} --output outserver1")
 server6 = Test.Processes.Process(
-    "server6", "bash -c '" + Test.TestDirectory + "/server1.sh {} outserver1'".format(Test.Variables.upstream_port6))
+    "server6", f"python3 {mock_origin} {Test.Variables.upstream_port6} {mock_origin_args} --output outserver1")
+server1.Ready = When.PortOpen(Test.Variables.upstream_port1)
+server2.Ready = When.PortOpen(Test.Variables.upstream_port2)
+server3.Ready = When.PortOpen(Test.Variables.upstream_port3)
+server4.Ready = When.PortOpen(Test.Variables.upstream_port4)
+server5.Ready = When.PortOpen(Test.Variables.upstream_port5)
+server6.Ready = When.PortOpen(Test.Variables.upstream_port6)
 
 big_post_body = "0123456789" * 231070
 big_post_body_file = open(os.path.join(Test.RunDirectory, "big_post_body"), "w")

tests/gold_tests/slow_post/server_abort.test.py

@@ -47,4 +47,5 @@
 tr.StillRunningAfter = server
 tr.StillRunningAfter = ts
 server.Streams.stderr += Testers.ContainsExpression(
-    "UnicodeDecodeError", "Verify that the server raises an exception when processing the request.")
+    "(UnicodeDecodeError|IndexError: list index out of range)",
+    "Verify that the server raises an exception when processing the request.")

tests/tools/mock_origin.py

@@ -0,0 +1,167 @@
+#!/usr/bin/env python3
+'''
+A reusable mock origin server for ATS autests.
+
+Replaces the various ad-hoc nc-based shell scripts (post/server1.sh,
+chunked_encoding/server2..4.sh, post_slow_server/server.sh) with a single
+Python tool that:
+
+  - Handles When.PortOpen() readiness probes gracefully (nc -l cannot).
+  - Accepts one real HTTP request, optionally saves it to a file, and sends
+    a configurable response.
+  - Drains remaining request data after responding so that ATS does not see
+    a connection reset while still forwarding a POST body (avoids HTTP/2 502).
+  - Supports Content-Length bodies, chunked transfer encoding, and arbitrary
+    response delays.
+
+Usage examples mapping to the original shell scripts:
+
+  # post/server1.sh PORT OUTFILE
+  mock_origin.py PORT --output OUTFILE --status 420 --reason "Be Calm"
+
+  # chunked_encoding/server2.sh PORT OUTFILE  (Content-Length body)
+  mock_origin.py PORT --output OUTFILE --body "123456789012345"
+
+  # chunked_encoding/server3.sh PORT OUTFILE  (Chunked body)
+  mock_origin.py PORT --output OUTFILE --body "123456789012345" --chunked
+
+  # post_slow_server/server.sh PORT  (Delayed 200KB response)
+  mock_origin.py PORT --output rcv_file --delay 120 --body-size 204800
+'''
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import argparse
+import socket
+import sys
+import time
+
+FILLER_LINE_WIDTH = 8
+
+
+def build_response(args):
+    '''Build the complete HTTP response bytes from CLI arguments.'''
+
+    body = b''
+    if args.body is not None:
+        body = args.body.encode()
+    elif args.body_size and args.body_size > 0:
+        lines = []
+        offset = 0
+        while offset < args.body_size:
+            offset += FILLER_LINE_WIDTH
+            lines.append(f'{offset:07d}\n'.encode())
+        body = b''.join(lines)[:args.body_size]
+
+    status_line = f'HTTP/1.1 {args.status} {args.reason}\r\n'.encode()
+
+    if args.chunked:
+        headers = b'Transfer-Encoding: chunked\r\n'
+        for h in (args.header or []):
+            headers += h.encode() + b'\r\n'
+        headers += b'\r\n'
+        chunk = f'{len(body):X}\r\n'.encode() + body + b'\r\n'
+        terminator = b'0\r\n\r\n'
+        return status_line + headers + chunk + terminator
+    else:
+        headers = f'Content-Length: {len(body)}\r\n'.encode()
+        for h in (args.header or []):
+            headers += h.encode() + b'\r\n'
+        headers += b'\r\n'
+        return status_line + headers + body
+
+
+def serve_one(args):
+    '''Listen, absorb readiness probes, serve one real HTTP transaction, exit.'''
+
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.bind(('', args.port))
+    sock.listen(1)
+
+    response = build_response(args)
+
+    while True:
+        conn, addr = sock.accept()
+        data = b''
+        try:
+            while True:
+                chunk = conn.recv(65536)
+                if not chunk:
+                    break
+                data += chunk
+                if b'\r\n\r\n' in data:
+                    break
+        except ConnectionError:
+            pass
+
+        if not data:
+            # Readiness probe (e.g. When.PortOpen) -- connected and
+            # disconnected without sending data.  Go back to waiting.
+            conn.close()
+            continue
+
+        # Real HTTP request arrived.
+        if args.output:
+            with open(args.output, 'wb') as f:
+                f.write(data)
+
+        if args.delay > 0:
+            time.sleep(args.delay)
+
+        try:
+            conn.sendall(response)
+        except ConnectionError:
+            pass
+
+        # Drain remaining request data (e.g. a large POST body that is still
+        # being forwarded by ATS).  Closing without draining causes a TCP RST
+        # which makes ATS return 502 on HTTP/2 streams.
+        try:
+            while True:
+                if not conn.recv(65536):
+                    break
+        except ConnectionError:
+            pass
+
+        conn.close()
+        break
+
+    sock.close()
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description='Mock origin server for ATS autests. '
+        'Listens on PORT, serves one HTTP transaction, then exits. '
+        'Compatible with When.PortOpen() readiness probes.')
+
+    parser.add_argument('port', type=int, help='TCP port to listen on')
+    parser.add_argument('--output', '-o', help='Write received request data to FILE')
+    parser.add_argument('--status', '-s', type=int, default=200, help='HTTP status code (default: 200)')
+    parser.add_argument('--reason', '-r', default='OK', help='HTTP reason phrase (default: OK)')
+    parser.add_argument('--header', action='append', help='Additional response header (repeatable), e.g. "X-Foo: bar"')
+    parser.add_argument('--body', '-b', help='Response body string')
+    parser.add_argument('--body-size', type=int, default=0, help='Generate N bytes of filler body data')
+    parser.add_argument('--chunked', action='store_true', help='Use chunked transfer encoding')
+    parser.add_argument('--delay', '-d', type=float, default=0, help='Seconds to delay before sending response')
+
+    args = parser.parse_args()
+    serve_one(args)
+
+
+if __name__ == '__main__':
+    main()