from __future__ import annotations import http.client import json import os import signal import socket import subprocess import sys import threading import textwrap import time from collections.abc import Callable from contextlib import contextmanager from pathlib import Path from typing import Iterator import pytest from aios_v2r1 import cli from aios_v2r1.ingress_service import ( MAX_RESPONSE_BYTES, MAX_REQUEST_BYTES, IngressHTTPServer, IngressRequestHandler, build_server, ) @contextmanager def running_server(processor: Callable[[str, bytes], tuple[int, dict[str, object]]] | None = None) -> Iterator[tuple[str, int]]: server = build_server("127.0.0.1", 0, processor=processor) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: host, port = server.server_address yield str(host), int(port) finally: server.shutdown() server.server_close() thread.join(timeout=5) def request( address: tuple[str, int], method: str, path: str, body: bytes | None = None, ) -> tuple[int, bytes, dict[str, str]]: connection = http.client.HTTPConnection(*address, timeout=5) try: connection.request(method, path, body=body) response = connection.getresponse() payload = response.read() return response.status, payload, {key.lower(): value for key, value in response.getheaders()} finally: connection.close() def raw_request( address: tuple[str, int], method: bytes, path: bytes = b"/health/live", ) -> bytes: request_bytes = ( method + b" " + path + b" HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n" ) with socket.create_connection(address, timeout=5) as connection: connection.sendall(request_bytes) chunks: list[bytes] = [] while True: chunk = connection.recv(8192) if not chunk: return b"".join(chunks) chunks.append(chunk) def raw_exchange( address: tuple[str, int], request_bytes: bytes, *, shutdown_write: bool = False, receive_timeout: float = 2.0, ) -> tuple[bytes, bool]: chunks: list[bytes] = [] timed_out = False with socket.create_connection(address, timeout=5) as connection: connection.settimeout(receive_timeout) connection.sendall(request_bytes) if shutdown_write: connection.shutdown(socket.SHUT_WR) while True: try: chunk = connection.recv(8192) except TimeoutError: timed_out = True break except ConnectionResetError: break if not chunk: break chunks.append(chunk) return b"".join(chunks), timed_out def parse_raw_response(response: bytes) -> tuple[int, bytes, bytes]: headers, body = response.split(b"\r\n\r\n", 1) status = int(headers.split(b"\r\n", 1)[0].split(b" ", 2)[1]) return status, headers, body def post_request_bytes(headers: bytes, body: bytes = b"") -> bytes: return ( b"POST /v1/ingress HTTP/1.1\r\n" b"Host: localhost\r\nConnection: close\r\n" + headers + b"\r\n" + body ) def wait_until_listening(process: subprocess.Popen[str], address: tuple[str, int]) -> None: deadline = time.monotonic() + 5 while time.monotonic() < deadline: if process.poll() is not None: _, stderr = process.communicate() pytest.fail(f"server exited before listening: {stderr}") try: with socket.create_connection(address, timeout=0.1): return except OSError: time.sleep(0.05) pytest.fail("server did not start listening") def test_live_and_unconfigured_readiness_are_exact_and_bounded() -> None: with running_server() as address: live = request(address, "GET", "/health/live") ready = request(address, "GET", "/health/ready") assert live[:2] == (200, b'{"status":"LIVE"}') assert ready[:2] == (503, b'{"status":"SERVICE_NOT_CONFIGURED"}') assert live[2]["content-type"] == "application/json" assert ready[2]["content-length"] == str(len(ready[1])) assert len(live[1]) <= 4096 assert len(ready[1]) <= 4096 @pytest.mark.parametrize( "route", ["/v1/ingress", "/v1/receipts/status", "/v1/outbound-receipts"], ) def test_mutation_routes_fail_closed_without_reading_or_mutating(route: str) -> None: with running_server() as address: status, payload, _ = request( address, "POST", route, b'{"sender":"do-not-log","body":"do-not-log"}', ) assert status == 503 assert payload == b'{"status":"SERVICE_NOT_CONFIGURED"}' def test_unknown_routes_and_wrong_methods_use_bounded_sanitized_codes() -> None: with running_server() as address: missing = request(address, "GET", "/missing") wrong_method = request(address, "GET", "/v1/ingress") unsupported = request(address, "PUT", "/health/live") assert missing[:2] == (404, b'{"status":"NOT_FOUND"}') assert wrong_method[:2] == (405, b'{"status":"METHOD_NOT_ALLOWED"}') assert unsupported[:2] == (405, b'{"status":"METHOD_NOT_ALLOWED"}') assert all(len(result[1]) <= 4096 for result in (missing, wrong_method, unsupported)) @pytest.mark.parametrize( ("method", "expected_status"), [ (b"METHOD_SECRET_REFLECTION_TOKEN", b"405"), (b"METHOD_SECRET_REFLECTION_TOKEN" + b"X" * 66_000, b"414"), ], ids=["arbitrary", "overlong"], ) def test_arbitrary_and_overlong_methods_use_bounded_json_without_reflection( method: bytes, expected_status: bytes ) -> None: with running_server() as address: response = raw_request(address, method) headers, body = response.split(b"\r\n\r\n", 1) status_line = headers.split(b"\r\n", 1)[0] assert status_line.split(b" ", 2)[1] == expected_status assert b"Content-Type: application/json" in headers assert len(body) <= 4096 assert b"METHOD_SECRET_REFLECTION_TOKEN" not in response assert b" None: def processor(route: str, body: bytes) -> tuple[int, dict[str, object]]: return invalid_status, {"status": "MUST_NOT_ESCAPE"} # type: ignore[return-value] with running_server(processor) as address: response = raw_request(address, b"POST", b"/v1/ingress") headers, body = response.split(b"\r\n\r\n", 1) assert headers.split(b"\r\n", 1)[0].startswith(b"HTTP/1.0 500 ") assert body == b'{"status":"INTERNAL_ERROR"}' assert b"MUST_NOT_ESCAPE" not in response def test_processor_is_injected_and_exceptions_are_not_disclosed(capsys: pytest.CaptureFixture[str]) -> None: secret = "sender-secret-body-token" def exploding_processor(route: str, body: bytes) -> tuple[int, dict[str, object]]: assert route == "/v1/ingress" assert json.loads(body) == {"value": secret} raise RuntimeError(secret) with running_server(exploding_processor) as address: ready = request(address, "GET", "/health/ready") failed = request( address, "POST", "/v1/ingress", json.dumps({"value": secret}).encode(), ) captured = capsys.readouterr() assert ready[:2] == (200, b'{"status":"READY"}') assert failed[:2] == (500, b'{"status":"INTERNAL_ERROR"}') assert secret not in failed[1].decode() assert secret not in captured.out assert secret not in captured.err @pytest.mark.parametrize( ("headers", "body"), [ (b"Transfer-Encoding: secret-framing-token\r\n", b"secret-body"), (b"Content-Length: 0\r\nContent-Length: 0\r\n", b""), (b"Content-Length: 1\r\nContent-Length: 2\r\n", b"x"), (b"Content-Length: invalid-secret-length\r\n", b""), (b"Content-Length: -1\r\n", b""), (f"Content-Length: {MAX_REQUEST_BYTES + 1}\r\n".encode(), b""), (b"Content-Length: 0 \r\n", b""), (b"Content-Length:\t0\r\n", b""), ], ids=[ "transfer-encoding", "duplicate-same", "duplicate-conflicting", "invalid", "negative", "oversize", "trailing-ows", "leading-ows", ], ) def test_invalid_post_framing_is_bounded_and_never_calls_processor( headers: bytes, body: bytes, capsys: pytest.CaptureFixture[str], ) -> None: calls: list[bytes] = [] def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]: calls.append(payload) return 201, {"status": "MUTATED"} with running_server(processor) as address: response, timed_out = raw_exchange( address, post_request_bytes(headers, body), ) status, response_headers, response_body = parse_raw_response(response) captured = capsys.readouterr() assert not timed_out assert 400 <= status <= 499 assert b"Content-Type: application/json" in response_headers assert len(response_body) <= 4096 assert b"secret-framing-token" not in response assert b"secret-body" not in response assert b"invalid-secret-length" not in response assert calls == [] assert "Traceback" not in captured.out assert "Traceback" not in captured.err def test_short_eof_body_is_rejected_before_processor() -> None: calls: list[bytes] = [] def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]: calls.append(payload) return 201, {"status": "MUTATED"} with running_server(processor) as address: response, timed_out = raw_exchange( address, post_request_bytes(b"Content-Length: 10\r\n", b"abc"), shutdown_write=True, ) status, _, body = parse_raw_response(response) assert not timed_out assert 400 <= status <= 499 assert len(body) <= 4096 assert calls == [] def test_partial_body_timeout_is_bounded_and_rejected_before_processor() -> None: processor_called = threading.Event() def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]: processor_called.set() return 201, {"status": "MUTATED"} with running_server(processor) as address: started = time.monotonic() response, timed_out = raw_exchange( address, post_request_bytes(b"Content-Length: 10\r\n", b"abc"), receive_timeout=1.5, ) elapsed = time.monotonic() - started called_after_client_close = processor_called.wait(timeout=0.5) status, _, body = parse_raw_response(response) assert not timed_out assert elapsed < 1.5 assert status == 408 assert len(body) <= 4096 assert not called_after_client_close def test_server_close_drains_partial_body_handler_and_prevents_late_processor( capsys: pytest.CaptureFixture[str], ) -> None: processor_called = threading.Event() def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]: processor_called.set() return 201, {"status": "MUTATED"} server = build_server("127.0.0.1", 0, processor=processor) server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() host, port = server.server_address client = socket.create_connection((str(host), int(port)), timeout=5) try: client.sendall(post_request_bytes(b"Content-Length: 10\r\n", b"abc")) time.sleep(0.1) started = time.monotonic() server.shutdown() server.server_close() close_elapsed = time.monotonic() - started finally: client.close() server_thread.join(timeout=5) called_after_close = processor_called.wait(timeout=0.5) captured = capsys.readouterr() assert server.daemon_threads is False assert server.block_on_close is True assert close_elapsed < 3 assert not server_thread.is_alive() assert not called_after_close assert "Traceback" not in captured.out assert "Traceback" not in captured.err assert "Fatal Python error" not in captured.err restarted = build_server(str(host), int(port)) restarted.server_close() @pytest.mark.parametrize( "target", [ b"http://[::1/v1/ingress", b"http://127.0.0.1:bad/v1/ingress", b"relative-target", b"/v1/ingress%ZZ", ], ids=[ "invalid-ipv6-bracket", "invalid-absolute-port", "invalid-origin-form", "invalid-percent-escape", ], ) def test_malformed_request_targets_return_sanitized_400_without_processor( target: bytes, capsys: pytest.CaptureFixture[str], ) -> None: calls: list[bytes] = [] def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]: calls.append(payload) return 201, {"status": "MUTATED"} request_bytes = ( b"POST " + target + b" HTTP/1.1\r\nHost: localhost\r\n" + b"Content-Length: 0\r\nConnection: close\r\n\r\n" ) with running_server(processor) as address: response, timed_out = raw_exchange(address, request_bytes) status, headers, body = parse_raw_response(response) captured = capsys.readouterr() assert not timed_out assert status == 400 assert b"Content-Type: application/json" in headers assert len(body) <= 4096 assert target not in response assert calls == [] assert "Traceback" not in captured.out assert "Traceback" not in captured.err def test_server_handle_error_never_prints_payload_traceback( capsys: pytest.CaptureFixture[str], ) -> None: secret = "unexpected-handler-secret" class ExplodingHandler(IngressRequestHandler): def do_GET(self) -> None: raise RuntimeError(secret) server = IngressHTTPServer(("127.0.0.1", 0), ExplodingHandler, None) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: host, port = server.server_address raw_exchange( (str(host), int(port)), b"GET /health/live HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", ) finally: server.shutdown() server.server_close() thread.join(timeout=5) captured = capsys.readouterr() assert secret not in captured.out assert secret not in captured.err assert "Traceback" not in captured.out assert "Traceback" not in captured.err @pytest.mark.parametrize( "malformed_header", [ b"Bad Name: value\r\n", b"Bad\tName: value\r\n", b"Trailing-Name : value\r\n", b" X-Leading: value\r\n", b"\tX-Leading: value\r\n", b":authority: localhost\r\n", b"X-Obs-Fold: first\r\n continuation\r\n", b"X-Obs-Fold: first\r\n\tcontinuation\r\n", b"X-Control: value\x00secret\r\n", b"X-Carriage: value\rsecret\r\n", b"X-Bare-LF: value\n", b"X-\xff: value\r\n", ], ids=[ "space-in-name", "tab-in-name", "whitespace-before-colon", "leading-space", "leading-tab", "pseudo-header", "obs-fold-space", "obs-fold-tab", "nul-in-value", "cr-in-value", "bare-lf", "non-ascii-name", ], ) def test_malformed_raw_header_grammar_is_rejected_before_processor( malformed_header: bytes, capsys: pytest.CaptureFixture[str], ) -> None: calls: list[bytes] = [] def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]: calls.append(payload) return 201, {"status": "MUTATED"} request_bytes = ( b"POST /v1/ingress HTTP/1.1\r\n" b"Host: localhost\r\n" + malformed_header + b"Content-Length: 0\r\nConnection: close\r\n\r\n" ) with running_server(processor) as address: response, timed_out = raw_exchange(address, request_bytes) assert not timed_out assert response status, headers, body = parse_raw_response(response) captured = capsys.readouterr() assert status == 400 assert b"Content-Type: application/json" in headers assert json.loads(body) == {"status": "BAD_REQUEST"} assert len(body) <= MAX_RESPONSE_BYTES assert b"secret" not in response assert calls == [] assert "Traceback" not in captured.out assert "Traceback" not in captured.err def trickle_request_until_response( address: tuple[str, int], prefix: bytes, *, receive_timeout: float = 1.8, ) -> tuple[bytes, bool, float]: connection = socket.create_connection(address, timeout=5) stop = threading.Event() def sender() -> None: while not stop.wait(0.1): try: connection.sendall(b"x") except OSError: return connection.sendall(prefix) sender_thread = threading.Thread(target=sender, daemon=True) sender_thread.start() connection.settimeout(receive_timeout) chunks: list[bytes] = [] timed_out = False started = time.monotonic() try: while True: try: chunk = connection.recv(8192) except TimeoutError: timed_out = True break except ConnectionResetError: break if not chunk: break chunks.append(chunk) finally: elapsed = time.monotonic() - started stop.set() try: connection.shutdown(socket.SHUT_RDWR) except OSError: pass connection.close() sender_thread.join(timeout=1) return b"".join(chunks), timed_out, elapsed @pytest.mark.parametrize( "prefix", [ ( b"POST /v1/ingress HTTP/1.1\r\n" b"Host: localhost\r\nX-Slow-Header: " ), ( b"POST /v1/ingress HTTP/1.1\r\n" b"Host: localhost\r\nContent-Length: 1000\r\n" b"Connection: close\r\n\r\nx" ), ], ids=["slow-header", "slow-body"], ) def test_absolute_network_deadline_cannot_be_reset_by_slow_trickle( prefix: bytes, ) -> None: processor_called = threading.Event() def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]: processor_called.set() return 201, {"status": "MUTATED"} with running_server(processor) as address: response, timed_out, elapsed = trickle_request_until_response(address, prefix) called_after_cleanup = processor_called.wait(timeout=0.4) assert not timed_out assert elapsed < 1.8 assert response status, _, body = parse_raw_response(response) assert status == 408 assert json.loads(body) == {"status": "REQUEST_TIMEOUT"} assert len(body) <= MAX_RESPONSE_BYTES assert not called_after_cleanup def test_shutdown_drains_several_slow_clients_without_late_work_or_threads( capsys: pytest.CaptureFixture[str], ) -> None: processor_called = threading.Event() def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]: processor_called.set() return 201, {"status": "MUTATED"} before_non_daemon = {id(thread) for thread in threading.enumerate() if not thread.daemon} server = build_server("127.0.0.1", 0, processor=processor) server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() host, port = server.server_address clients: list[socket.socket] = [] stops: list[threading.Event] = [] senders: list[threading.Thread] = [] def start_client(prefix: bytes) -> None: client = socket.create_connection((str(host), int(port)), timeout=5) client.sendall(prefix) stop = threading.Event() def sender() -> None: while not stop.wait(0.1): try: client.sendall(b"x") except OSError: return sender_thread = threading.Thread(target=sender, daemon=True) sender_thread.start() clients.append(client) stops.append(stop) senders.append(sender_thread) header_prefix = ( b"POST /v1/ingress HTTP/1.1\r\nHost: localhost\r\nX-Slow: " ) body_prefix = ( b"POST /v1/ingress HTTP/1.1\r\nHost: localhost\r\n" b"Content-Length: 1000\r\nConnection: close\r\n\r\nx" ) for index in range(6): start_client(header_prefix if index % 2 == 0 else body_prefix) time.sleep(0.15) closed = threading.Event() close_elapsed: list[float] = [] def close_server() -> None: started = time.monotonic() try: server.shutdown() server.server_close() finally: close_elapsed.append(time.monotonic() - started) closed.set() closer = threading.Thread(target=close_server) closer.start() closed_within_bound = closed.wait(timeout=2.5) for stop in stops: stop.set() for client in clients: try: client.shutdown(socket.SHUT_RDWR) except OSError: pass client.close() for sender_thread in senders: sender_thread.join(timeout=1) closer.join(timeout=5) server_thread.join(timeout=5) called_after_close = processor_called.wait(timeout=0.4) deadline = time.monotonic() + 1 leaked_non_daemon: list[threading.Thread] = [] while time.monotonic() < deadline: leaked_non_daemon = [ thread for thread in threading.enumerate() if not thread.daemon and id(thread) not in before_non_daemon and thread.is_alive() ] if not leaked_non_daemon: break time.sleep(0.02) captured = capsys.readouterr() assert closed_within_bound assert close_elapsed and close_elapsed[0] < 2.5 assert not server_thread.is_alive() assert not closer.is_alive() assert not leaked_non_daemon assert not called_after_close assert "Traceback" not in captured.out assert "Traceback" not in captured.err assert "Fatal Python error" not in captured.err restarted = build_server(str(host), int(port)) restarted.server_close() def test_completed_framing_cancels_deadline_before_slow_processor() -> None: def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]: time.sleep(1.2) return 201, {"status": "MUTATED"} with running_server(processor) as address: status, body, _ = request(address, "POST", "/v1/ingress", b"") assert status == 201 assert json.loads(body) == {"status": "MUTATED"} assert not [ thread for thread in threading.enumerate() if thread.name.startswith("aios-ingress-read-deadline") and thread.is_alive() ] @pytest.mark.parametrize( "payload_factory", [ pytest.param(lambda: [], id="list"), pytest.param(lambda: "not-an-object", id="string"), pytest.param(lambda: None, id="null"), pytest.param(lambda: {"value": float("nan")}, id="nan"), pytest.param(lambda: {"value": float("inf")}, id="infinity"), pytest.param(lambda: {"value": float("-inf")}, id="negative-infinity"), pytest.param(lambda: {"value": object()}, id="nonserializable"), pytest.param( lambda: {"value": "secret-oversize-" + "x" * MAX_RESPONSE_BYTES}, id="oversize", ), ], ) def test_invalid_processor_payload_is_fixed_bounded_json_500( payload_factory: Callable[[], object], capsys: pytest.CaptureFixture[str], ) -> None: payload = payload_factory() def processor(route: str, body: bytes) -> tuple[int, dict[str, object]]: return 201, payload # type: ignore[return-value] with running_server(processor) as address: response = raw_request(address, b"POST", b"/v1/ingress") status, headers, body = parse_raw_response(response) captured = capsys.readouterr() assert status == 500 assert b"Content-Type: application/json" in headers assert body == b'{"status":"INTERNAL_ERROR"}' assert json.loads(body) == {"status": "INTERNAL_ERROR"} assert len(body) <= MAX_RESPONSE_BYTES assert b"secret-oversize" not in response assert "Traceback" not in captured.out assert "Traceback" not in captured.err def test_recursive_processor_payload_uses_nonrecursive_fixed_500() -> None: recursive: dict[str, object] = {} recursive["self"] = recursive def processor(route: str, body: bytes) -> tuple[int, dict[str, object]]: return 201, recursive with running_server(processor) as address: status, body, _ = request(address, "POST", "/v1/ingress", b"") assert status == 500 assert body == b'{"status":"INTERNAL_ERROR"}' assert json.loads(body) == {"status": "INTERNAL_ERROR"} def test_server_rejects_non_loopback_bind() -> None: with pytest.raises(ValueError, match="loopback"): build_server("0.0.0.0", 0) def test_af_inet_server_rejects_ipv6_loopback_explicitly() -> None: with pytest.raises(ValueError, match="IPv4 loopback"): build_server("::1", 0) def run_pre_header_sigint_probe( timeout: float = 1.2, ) -> tuple[int, str, str, float, bool, int]: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as reservation: reservation.bind(("127.0.0.1", 0)) port = int(reservation.getsockname()[1]) script = textwrap.dedent( """ import os import signal import socket import sys import threading import time from aios_v2r1.ingress_service import ( IngressHTTPServer, IngressRequestHandler, ) accepted_before_setup = threading.Event() class PreHeaderReadinessHandler(IngressRequestHandler): def setup(self): accepted_before_setup.set() try: self.request.recv(1) except OSError: pass super().setup() server = IngressHTTPServer( ("127.0.0.1", int(sys.argv[1])), PreHeaderReadinessHandler, None, ) def connect_and_interrupt(): client = socket.create_connection(server.server_address, timeout=2) if not accepted_before_setup.wait(timeout=2): os._exit(71) os.kill(os.getpid(), signal.SIGINT) time.sleep(10) client.close() trigger = threading.Thread(target=connect_and_interrupt, daemon=True) trigger.start() try: server.serve_forever(poll_interval=0.01) except KeyboardInterrupt: pass finally: server.server_close() """ ) env = os.environ.copy() package_root = str(Path(__file__).resolve().parents[2] / "active-runtime" / "app") env["PYTHONPATH"] = package_root + os.pathsep + env.get("PYTHONPATH", "") process = subprocess.Popen( [sys.executable, "-c", script, str(port)], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) started = time.monotonic() timed_out = False try: stdout, stderr = process.communicate(timeout=timeout) except subprocess.TimeoutExpired: timed_out = True process.kill() stdout, stderr = process.communicate(timeout=2) return ( int(process.returncode), stdout, stderr, time.monotonic() - started, timed_out, port, ) def test_sigint_closes_pre_header_readiness_socket_in_25_bounded_runs() -> None: for iteration in range(25): returncode, stdout, stderr, elapsed, timed_out, port = ( run_pre_header_sigint_probe() ) assert not timed_out, f"pre-header SIGINT hung on iteration {iteration}" assert elapsed < 1.2 assert returncode == 0 assert "Traceback" not in stdout assert "Traceback" not in stderr assert "Fatal Python error" not in stderr restarted = build_server("127.0.0.1", port) restarted.server_close() def test_server_close_closes_socket_accepted_before_handler_setup() -> None: server = build_server("127.0.0.1", 0) client = socket.create_connection(server.server_address, timeout=2) accepted, _ = server.get_request() client.settimeout(0.4) try: server.server_close() server.server_close() try: remaining = client.recv(1) except ConnectionResetError: remaining = b"" assert remaining == b"" finally: accepted.close() client.close() server.server_close() def test_process_request_rejects_socket_after_server_closing() -> None: handler_started = threading.Event() class RecordingHandler: def __init__(self, *args: object) -> None: handler_started.set() server = IngressHTTPServer( ("127.0.0.1", 0), RecordingHandler, # type: ignore[arg-type] None, ) server.server_close() request_socket, peer = socket.socketpair() peer.settimeout(0.4) try: server.process_request(request_socket, ("127.0.0.1", 1)) try: remaining = peer.recv(1) except ConnectionResetError: remaining = b"" assert remaining == b"" assert not handler_started.wait(timeout=0.1) finally: request_socket.close() peer.close() server.server_close() def test_accept_and_server_close_race_is_bounded_in_25_runs() -> None: errors: list[BaseException] = [] for _ in range(25): class RecordingHandler: def __init__(self, *args: object) -> None: return server = IngressHTTPServer( ("127.0.0.1", 0), RecordingHandler, # type: ignore[arg-type] None, ) client = socket.create_connection(server.server_address, timeout=2) barrier = threading.Barrier(3) def accept_and_process() -> None: try: barrier.wait(timeout=1) request_socket, client_address = server.get_request() server.process_request(request_socket, client_address) except OSError: return except BaseException as exc: errors.append(exc) def close_during_accept() -> None: try: barrier.wait(timeout=1) server.server_close() except BaseException as exc: errors.append(exc) accepter = threading.Thread(target=accept_and_process) closer = threading.Thread(target=close_during_accept) accepter.start() closer.start() barrier.wait(timeout=1) accepter.join(timeout=1) closer.join(timeout=1) client.close() server.server_close() assert not accepter.is_alive() assert not closer.is_alive() assert not errors def test_server_close_is_idempotent_with_idle_and_partial_clients( capsys: pytest.CaptureFixture[str], ) -> None: server = build_server("127.0.0.1", 0) server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() clients: list[socket.socket] = [] payloads = [ None, None, b"GET /health/live HTTP/1.1\r\nHost: localhost\r\n", b"POST /v1/ingress HTTP/1.1\r\nContent-Length: 10\r\n\r\nx", ] for payload in payloads: client = socket.create_connection(server.server_address, timeout=2) if payload is not None: client.sendall(payload) clients.append(client) time.sleep(0.05) server.shutdown() errors: list[BaseException] = [] def close_server() -> None: try: server.server_close() except BaseException as exc: errors.append(exc) closers = [threading.Thread(target=close_server) for _ in range(4)] for closer in closers: closer.start() for closer in closers: closer.join(timeout=2) server_thread.join(timeout=2) for client in clients: client.close() captured = capsys.readouterr() assert not errors assert not [closer for closer in closers if closer.is_alive()] assert not server_thread.is_alive() assert "Traceback" not in captured.out assert "Traceback" not in captured.err assert "Fatal Python error" not in captured.err def test_sigint_shutdown_exits_zero_without_traceback_and_releases_port() -> None: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as reservation: reservation.bind(("127.0.0.1", 0)) port = int(reservation.getsockname()[1]) env = os.environ.copy() package_root = str(Path(__file__).resolve().parents[2] / "active-runtime" / "app") env["PYTHONPATH"] = package_root + os.pathsep + env.get("PYTHONPATH", "") process = subprocess.Popen( [ sys.executable, "-m", "aios_v2r1.cli", "serve", "--host", "127.0.0.1", "--port", str(port), ], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) try: wait_until_listening(process, ("127.0.0.1", port)) process.send_signal(signal.SIGINT) stdout, stderr = process.communicate(timeout=5) finally: if process.poll() is None: process.kill() process.communicate() assert process.returncode == 0 assert "Traceback" not in stdout assert "Traceback" not in stderr with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as rebound: rebound.bind(("127.0.0.1", port)) def test_cli_preserves_registry_commands_and_dispatches_top_level_serve(monkeypatch: pytest.MonkeyPatch) -> None: parsed = cli.build_parser().parse_args(["registry.sqlite", "status", "tenant-a"]) assert parsed.registry == "registry.sqlite" assert parsed.command == "status" called: list[tuple[str, int]] = [] monkeypatch.setattr(cli, "serve", lambda host, port: called.append((host, port))) assert cli.main(["serve", "--host", "127.0.0.1", "--port", "19001"]) == 0 assert called == [("127.0.0.1", 19001)]