From 8b426c18cc7be82aaac1f08c5978d74313f561d2 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 00:41:02 -0500 Subject: [PATCH] feat(kernel): add passive TCP server abstractions --- README.md | 9 + SmallPackage/Kernel.py | 226 +++++++++- SmallPackage/_types.py | 17 + demos/web_app_demo.py | 78 ++-- tests/test_server_kernel.py | 431 ++++++++++++++++++++ tests/typing/passive_tcp_kernel_contract.py | 10 + 6 files changed, 740 insertions(+), 31 deletions(-) create mode 100644 tests/test_server_kernel.py create mode 100644 tests/typing/passive_tcp_kernel_contract.py diff --git a/README.md b/README.md index 22045ac..f3b5caf 100644 --- a/README.md +++ b/README.md @@ -460,6 +460,15 @@ The kernel layer is deliberately generic. Protocol clients such as HTTPS, Redis, MQTT, RabbitMQ/AMQP, and Kafka should be built on top of the shared TCP/TLS socket surface rather than requiring protocol-specific kernel methods. +Passive TCP consumers use the kernel boundary as well: check +`supports_tcp_server()` before resolving or opening anything, pass the opaque +record returned by `resolve_passive_address()` unchanged to both `socket_open()` +and `socket_bind()`, then use the kernel's listen, accept, address-inspection, +and close operations. Address reuse has its own capability check because some +MicroPython ports support listeners without exposing `SO_REUSEADDR` constants. +The web app demo shows the complete setup and rollback pattern without importing +platform socket APIs. + ## Clients The current setup now includes first-party smallOS-native helpers for HTTP, diff --git a/SmallPackage/Kernel.py b/SmallPackage/Kernel.py index e01bed6..f793889 100644 --- a/SmallPackage/Kernel.py +++ b/SmallPackage/Kernel.py @@ -29,6 +29,21 @@ _UNSET = object() +class _PassiveAddressInfo: + """Kernel-owned address record for a passive TCP endpoint.""" + + def __init__(self, owner, address_info): + self.owner = owner + self.address_info = address_info + + +def _unwrap_passive_address(owner, address_info): + """Return backend address data only for a record created by ``owner``.""" + if not isinstance(address_info, _PassiveAddressInfo) or address_info.owner is not owner: + raise ValueError('Passive address record belongs to a different kernel.') + return address_info.address_info + + def _import_first(*module_names: str) -> Any | None: """ Import the first available module name from a list. @@ -512,6 +527,14 @@ def supports_external_wait_objects(self) -> bool: """Whether ``io_wait`` can wake on adapter-owned readiness objects.""" return False + def supports_tcp_server(self) -> bool: + """Whether this kernel implements the complete passive TCP contract.""" + return False + + def supports_reuse_address(self) -> bool: + """Whether this kernel can configure address reuse on a listener.""" + return False + def validate_io_wait_object( self, obj: Any ) -> tuple[bool, BaseException | None]: @@ -544,11 +567,32 @@ def validate_io_wait_object( def resolve_address(self, host: str, port: int) -> Any: return None + def resolve_passive_address(self, host: str, port: int) -> Any: + raise NotImplementedError('Passive TCP address resolution is not supported.') + def socket_open(self, address_info: Any) -> Any: - return None + raise NotImplementedError('TCP sockets are not supported.') def socket_setblocking(self, sock: Any, flag: bool) -> None: - return + raise NotImplementedError('TCP sockets are not supported.') + + def socket_set_reuse_address(self, sock: Any, enabled: bool) -> None: + raise NotImplementedError('TCP address reuse is not supported.') + + def socket_bind(self, sock: Any, address_info: Any) -> None: + raise NotImplementedError('TCP listeners are not supported.') + + def socket_listen(self, sock: Any, backlog: int) -> None: + raise NotImplementedError('TCP listeners are not supported.') + + def socket_accept(self, listener: Any) -> tuple[Any, Any]: + raise NotImplementedError('TCP listeners are not supported.') + + def socket_local_address(self, sock: Any) -> Any: + raise NotImplementedError('Socket address inspection is not supported.') + + def socket_peer_address(self, sock: Any) -> Any | None: + raise NotImplementedError('Socket address inspection is not supported.') def socket_connect(self, sock: Any, sockaddr: Any) -> bool: return True @@ -563,7 +607,7 @@ def socket_recv(self, sock: Any, buffer_size: int) -> bytes: return b'' def socket_close(self, sock: Any) -> None: - return + raise NotImplementedError('TCP sockets are not supported.') def socket_wrap_tls_client( self, @@ -736,10 +780,32 @@ def validate_io_wait_object(self, obj: Any) -> tuple[bool, BaseException | None] def supports_external_wait_objects(self) -> bool: return True + def supports_tcp_server(self) -> bool: + return True + + def supports_reuse_address(self) -> bool: + return ( + hasattr(self._socket, 'SOL_SOCKET') + and hasattr(self._socket, 'SO_REUSEADDR') + and callable(getattr(self._socket.socket, 'setsockopt', None)) + ) + def resolve_address(self, host, port): return self._socket.getaddrinfo(host, port, type=self._socket.SOCK_STREAM)[0] + def resolve_passive_address(self, host, port): + passive_host = None if host == '' else host + address_info = self._socket.getaddrinfo( + passive_host, + port, + type=self._socket.SOCK_STREAM, + flags=self._socket.AI_PASSIVE, + )[0] + return _PassiveAddressInfo(self, address_info) + def socket_open(self, address_info): + if isinstance(address_info, _PassiveAddressInfo): + address_info = _unwrap_passive_address(self, address_info) family, socktype, proto, _, _ = address_info return self._socket.socket(family, socktype, proto) @@ -747,6 +813,39 @@ def socket_setblocking(self, sock, flag): sock.setblocking(flag) return + def socket_set_reuse_address(self, sock, enabled): + if not self.supports_reuse_address(): + raise NotImplementedError('SO_REUSEADDR is not supported on this platform.') + sock.setsockopt( + self._socket.SOL_SOCKET, + self._socket.SO_REUSEADDR, + 1 if enabled else 0, + ) + return + + def socket_bind(self, sock, address_info): + address_info = _unwrap_passive_address(self, address_info) + sock.bind(address_info[4]) + return + + def socket_listen(self, sock, backlog): + sock.listen(backlog) + return + + def socket_accept(self, listener): + return listener.accept() + + def socket_local_address(self, sock): + return sock.getsockname() + + def socket_peer_address(self, sock): + try: + return sock.getpeername() + except OSError as exc: + if self._extract_errno(exc) == self._errno.ENOTCONN: + return None + raise + def socket_connect(self, sock, sockaddr): err = sock.connect_ex(sockaddr) pending = { @@ -848,6 +947,7 @@ def __init__( self._ssl = _import_first('ssl', 'ussl') self._sys = modules.get('sys') or _import_first('sys') self._os = modules.get('os') or _import_first('os') + self._errno = modules.get('errno') or _import_first('errno', 'uerrno') self._network = modules.get('network') or _import_first('network') self._rp2 = modules.get('rp2') or _import_first('rp2') self._machine = modules.get('machine') or _import_first('machine') @@ -936,17 +1036,135 @@ def create_io_wait_set(self): def supports_external_wait_objects(self) -> bool: return bool(self._poll_factory) + def supports_tcp_server(self) -> bool: + if not callable(getattr(self._socket, 'getaddrinfo', None)): + return False + socket_factory = getattr(self._socket, 'socket', None) + if not callable(socket_factory): + return False + required = ( + 'setblocking', 'bind', 'listen', 'accept', 'getsockname', + 'getpeername', 'send', 'recv', 'close', + ) + return all(callable(getattr(socket_factory, name, None)) for name in required) + + def supports_reuse_address(self) -> bool: + socket_factory = getattr(self._socket, 'socket', None) + return ( + hasattr(self._socket, 'SOL_SOCKET') + and hasattr(self._socket, 'SO_REUSEADDR') + and callable(getattr(socket_factory, 'setsockopt', None)) + ) + def resolve_address(self, host, port): return self._socket.getaddrinfo(host, port)[0] + def resolve_passive_address(self, host, port): + if not self.supports_tcp_server(): + raise NotImplementedError('TCP server support is not available on this port.') + passive_host = '0.0.0.0' if host == '' else host + sock_stream = getattr(self._socket, 'SOCK_STREAM', None) + ai_passive = getattr(self._socket, 'AI_PASSIVE', None) + attempts = [] + if sock_stream is not None and ai_passive is not None: + attempts.append((passive_host, port, 0, sock_stream, 0, ai_passive)) + if sock_stream is not None: + attempts.append((passive_host, port, 0, sock_stream)) + attempts.append((passive_host, port)) + last_error = None + for arguments in attempts: + try: + resolved = self._socket.getaddrinfo(*arguments)[0] + break + except TypeError as exc: + last_error = exc + else: + if last_error is None: + raise RuntimeError('Passive TCP address resolution did not run.') + raise last_error + return _PassiveAddressInfo(self, resolved) + def socket_open(self, address_info): + is_passive = isinstance(address_info, _PassiveAddressInfo) + if is_passive: + address_info = _unwrap_passive_address(self, address_info) family, socktype, proto, _, _ = address_info - return self._socket.socket(family, socktype, proto) + sock = self._socket.socket(family, socktype, proto) + if not is_passive: + return sock + required = ( + 'setblocking', 'bind', 'listen', 'accept', 'getsockname', + 'getpeername', 'send', 'recv', 'close', + ) + missing = [name for name in required if not callable(getattr(sock, name, None))] + if not missing: + return sock + closer = getattr(sock, 'close', None) + if callable(closer): + try: + closer() + except BaseException: + pass + raise NotImplementedError( + 'TCP server socket is missing required operations: {}.'.format(', '.join(missing)) + ) def socket_setblocking(self, sock, flag): sock.setblocking(flag) return + def socket_set_reuse_address(self, sock, enabled): + if not self.supports_reuse_address(): + raise NotImplementedError('SO_REUSEADDR is not supported on this port.') + setter = getattr(sock, 'setsockopt', None) + if not callable(setter): + raise NotImplementedError('Socket instance does not support SO_REUSEADDR.') + setter( + self._socket.SOL_SOCKET, + self._socket.SO_REUSEADDR, + 1 if enabled else 0, + ) + return + + def socket_bind(self, sock, address_info): + address_info = _unwrap_passive_address(self, address_info) + binder = getattr(sock, 'bind', None) + if not callable(binder): + raise NotImplementedError('TCP bind is not supported on this port.') + binder(address_info[4]) + return + + def socket_listen(self, sock, backlog): + listener = getattr(sock, 'listen', None) + if not callable(listener): + raise NotImplementedError('TCP listen is not supported on this port.') + listener(backlog) + return + + def socket_accept(self, listener): + acceptor = getattr(listener, 'accept', None) + if not callable(acceptor): + raise NotImplementedError('TCP accept is not supported on this port.') + return acceptor() + + def socket_local_address(self, sock): + getter = getattr(sock, 'getsockname', None) + if not callable(getter): + raise NotImplementedError('Local address inspection is not supported.') + return getter() + + def socket_peer_address(self, sock): + getter = getattr(sock, 'getpeername', None) + if not callable(getter): + raise NotImplementedError('Peer address inspection is not supported.') + try: + return getter() + except OSError as exc: + not_connected = getattr(self._errno, 'ENOTCONN', None) + if not_connected is not None and self._extract_errno(exc) == not_connected: + return None + raise + def socket_connect(self, sock, sockaddr): try: sock.connect(sockaddr) diff --git a/SmallPackage/_types.py b/SmallPackage/_types.py index 6589d5e..3b59862 100644 --- a/SmallPackage/_types.py +++ b/SmallPackage/_types.py @@ -42,6 +42,23 @@ def io_wait( ) -> tuple[Sequence[Any], Sequence[Any]]: ... +class PassiveTCPKernelLike(Protocol): + """Platform-neutral passive TCP operations used by server consumers.""" + + def supports_tcp_server(self) -> bool: ... + def supports_reuse_address(self) -> bool: ... + def resolve_passive_address(self, host: str, port: int) -> Any: ... + def socket_open(self, address_info: Any) -> Any: ... + def socket_setblocking(self, sock: Any, flag: bool) -> None: ... + def socket_set_reuse_address(self, sock: Any, enabled: bool) -> None: ... + def socket_bind(self, sock: Any, address_info: Any) -> None: ... + def socket_listen(self, sock: Any, backlog: int) -> None: ... + def socket_accept(self, listener: Any) -> tuple[Any, Any]: ... + def socket_local_address(self, sock: Any) -> Any: ... + def socket_peer_address(self, sock: Any) -> Any | None: ... + def socket_close(self, sock: Any) -> None: ... + + class AdapterCompletionLike(Protocol): """Structural completion record consumed by the scheduler.""" diff --git a/demos/web_app_demo.py b/demos/web_app_demo.py index 6244b30..3d2595e 100644 --- a/demos/web_app_demo.py +++ b/demos/web_app_demo.py @@ -340,34 +340,64 @@ async def metrics_task(task, state): await task.sleep(METRICS_INTERVAL_SECONDS) +def _close_after_setup_failure(kernel, stream): + """Best-effort rollback that never replaces the setup exception.""" + try: + kernel.socket_close(stream) + except BaseException: + pass + + +def _open_listener(kernel, host, port, backlog): + """Construct one listener entirely through the passive TCP kernel contract.""" + if not kernel.supports_tcp_server(): + raise NotImplementedError("This kernel does not support passive TCP servers.") + + address_info = kernel.resolve_passive_address(host, port) + listener = kernel.socket_open(address_info) + try: + if kernel.supports_reuse_address(): + kernel.socket_set_reuse_address(listener, True) + kernel.socket_bind(listener, address_info) + kernel.socket_listen(listener, backlog) + kernel.socket_setblocking(listener, False) + except BaseException: + _close_after_setup_failure(kernel, listener) + raise + return listener + + +def _dispatch_client(task, client_sock, client_addr, state): + """Transfer one accepted stream to a handler or roll it back exactly once.""" + kernel = task.OS.kernel + try: + kernel.socket_setblocking(client_sock, False) + task.spawn( + web_client_handler, + priority=max(1, task.priority - 1), + name="http_client", + args=(client_sock, client_addr, state), + ) + except BaseException: + _close_after_setup_failure(kernel, client_sock) + raise + + async def web_server_task(task, state): """Run a small cooperative HTTP server on one non-blocking listener.""" kernel = task.OS.kernel if kernel is None: raise RuntimeError("web_server_task requires a kernel-enabled runtime.") - address_info = kernel.resolve_address(HOST, PORT) - listener = kernel.socket_open(address_info) - - # Reuse address when available so restarting the demo is less annoying. - if hasattr(listener, "setsockopt") and hasattr(listener, "SOL_SOCKET") and hasattr(listener, "SO_REUSEADDR"): - try: - listener.setsockopt(listener.SOL_SOCKET, listener.SO_REUSEADDR, 1) - except Exception: - pass - - listener.bind(address_info[4]) - listener.listen(LISTEN_BACKLOG) - kernel.socket_setblocking(listener, False) - - task.OS.print("smallOS web app running on http://{}:{}/\n".format(HOST, PORT)) - task.OS.print("routes: / /api/stats /api/time /healthz\n") - task.OS.print("web_server PID: {}\n".format(task.getID())) - + listener = _open_listener(kernel, HOST, PORT, LISTEN_BACKLOG) try: + task.OS.print("smallOS web app running on http://{}:{}/\n".format(HOST, PORT)) + task.OS.print("routes: / /api/stats /api/time /healthz\n") + task.OS.print("web_server PID: {}\n".format(task.getID())) + while True: try: - client_sock, client_addr = listener.accept() + client_sock, client_addr = kernel.socket_accept(listener) except Exception as exc: if kernel.socket_needs_read(exc): await task.wait_readable(listener) @@ -377,16 +407,10 @@ async def web_server_task(task, state): continue raise - kernel.socket_setblocking(client_sock, False) - task.spawn( - web_client_handler, - priority=max(1, task.priority - 1), - name="http_client", - args=(client_sock, client_addr, state), - ) + _dispatch_client(task, client_sock, client_addr, state) await task.yield_now() finally: - kernel.socket_close(listener) + _close_after_setup_failure(kernel, listener) def main(): diff --git a/tests/test_server_kernel.py b/tests/test_server_kernel.py new file mode 100644 index 0000000..42fa25e --- /dev/null +++ b/tests/test_server_kernel.py @@ -0,0 +1,431 @@ +import asyncio +import os +import socket +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "demos")) + +from demos.web_app_demo import _dispatch_client, _open_listener, web_server_task +from SmallPackage.Kernel import Kernel, MicroPythonKernel, Unix + + +class FakeStream: + def __init__(self): + self.blocking_calls = [] + self.option_calls = [] + self.bind_calls = [] + self.listen_calls = [] + self.accept_result = (object(), ("192.0.2.2", 5000)) + self.local_address = ("0.0.0.0", 8080) + self.peer_address = ("192.0.2.2", 5000) + self.peer_error = None + self.close_calls = 0 + + def setblocking(self, flag): + self.blocking_calls.append(flag) + + def setsockopt(self, level, option, value): + self.option_calls.append((level, option, value)) + + def bind(self, address): + self.bind_calls.append(address) + + def listen(self, backlog): + self.listen_calls.append(backlog) + + def accept(self): + return self.accept_result + + def getsockname(self): + return self.local_address + + def getpeername(self): + if self.peer_error is not None: + raise self.peer_error + return self.peer_address + + def send(self, data): + return len(data) + + def recv(self, _size): + return b"" + + def close(self): + self.close_calls += 1 + + +class FakeSocketFactory: + setblocking = FakeStream.setblocking + setsockopt = FakeStream.setsockopt + bind = FakeStream.bind + listen = FakeStream.listen + accept = FakeStream.accept + getsockname = FakeStream.getsockname + getpeername = FakeStream.getpeername + send = FakeStream.send + recv = FakeStream.recv + close = FakeStream.close + + def __init__(self, module): + self.module = module + + def __call__(self, family, socktype, proto): + self.module.socket_calls.append((family, socktype, proto)) + return self.module.stream + + +class TwoArgumentSocketModule: + """MicroPython-style module exposing only a narrow resolver signature.""" + + SOCK_STREAM = 1 + + def __init__(self, stream): + self.stream = stream + self.resolve_calls = [] + self.socket_calls = [] + self.socket = FakeSocketFactory(self) + + def getaddrinfo(self, *args): + self.resolve_calls.append(args) + if len(args) != 2: + raise TypeError("this port accepts only host and port") + host, port = args + return [(2, self.SOCK_STREAM, 6, "", (host, port))] + + +class MissingServerSocketModule: + SOCK_STREAM = 1 + + +class IncompleteStream: + def __init__(self): + self.close_calls = 0 + + def close(self): + self.close_calls += 1 + + +class RecordingServerKernel: + def __init__(self, fail_at=None, close_fails=False, supported=True): + self.fail_at = fail_at + self.close_fails = close_fails + self.supported = supported + self.calls = [] + self.record = object() + self.listener = object() + + def _call(self, name): + self.calls.append(name) + if self.fail_at == name: + raise RuntimeError("{} failed".format(name)) + + def supports_tcp_server(self): + self.calls.append("supports_tcp_server") + return self.supported + + def supports_reuse_address(self): + self.calls.append("supports_reuse_address") + return True + + def resolve_passive_address(self, host, port): + self._call("resolve_passive_address") + self.resolved = (host, port) + return self.record + + def socket_open(self, record): + self._call("socket_open") + self.open_record = record + return self.listener + + def socket_set_reuse_address(self, listener, enabled): + self._call("socket_set_reuse_address") + self.reuse_args = (listener, enabled) + + def socket_bind(self, listener, record): + self._call("socket_bind") + self.bind_args = (listener, record) + + def socket_listen(self, listener, backlog): + self._call("socket_listen") + self.listen_args = (listener, backlog) + + def socket_setblocking(self, listener, flag): + self._call("socket_setblocking") + self.blocking_args = (listener, flag) + + def socket_accept(self, listener): + self._call("socket_accept") + return object(), ("192.0.2.1", 5000) + + def socket_needs_read(self, exc): + return False + + def socket_needs_write(self, exc): + return False + + def socket_close(self, listener): + self.calls.append("socket_close") + self.closed_listener = listener + if self.close_fails: + raise RuntimeError("close failed") + + +class DispatchTask: + priority = 2 + + def __init__(self, kernel, spawn_fails=False): + self.OS = type("OS", (), {"kernel": kernel})() + self.spawn_fails = spawn_fails + self.spawn_calls = [] + + def spawn(self, routine, **kwargs): + self.spawn_calls.append((routine, kwargs)) + if self.spawn_fails: + raise RuntimeError("spawn failed") + + +class ServerTask(DispatchTask): + def __init__(self, kernel, output_fails=False): + super().__init__(kernel) + self.output_fails = output_fails + self.OS.print = self._print + + def _print(self, _message): + if self.output_fails: + raise RuntimeError("output failed") + + def getID(self): + return 1 + + +class TestServerKernelContract(unittest.TestCase): + def test_base_kernel_reports_unsupported_passive_operations(self): + kernel = Kernel() + + self.assertFalse(kernel.supports_tcp_server()) + self.assertFalse(kernel.supports_reuse_address()) + unsupported_calls = ( + lambda: kernel.resolve_passive_address("", 0), + lambda: kernel.socket_open(object()), + lambda: kernel.socket_setblocking(object(), False), + lambda: kernel.socket_set_reuse_address(object(), True), + lambda: kernel.socket_bind(object(), object()), + lambda: kernel.socket_listen(object(), 1), + lambda: kernel.socket_accept(object()), + lambda: kernel.socket_local_address(object()), + lambda: kernel.socket_peer_address(object()), + lambda: kernel.socket_close(object()), + ) + for call in unsupported_calls: + with self.subTest(call=call), self.assertRaises(NotImplementedError): + call() + + def test_unix_loopback_echo_uses_only_kernel_operations(self): + kernel = Unix() + record = kernel.resolve_passive_address("127.0.0.1", 0) + listener = kernel.socket_open(record) + client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + accepted = None + try: + self.assertTrue(kernel.supports_tcp_server()) + self.assertNotIsInstance(record, tuple) + if kernel.supports_reuse_address(): + kernel.socket_set_reuse_address(listener, True) + try: + kernel.socket_bind(listener, record) + except PermissionError as exc: + self.skipTest("loopback bind is not permitted: {}".format(exc)) + kernel.socket_listen(listener, 2) + local_address = kernel.socket_local_address(listener) + self.assertGreater(local_address[1], 0) + + client.connect(local_address) + accepted, accepted_address = kernel.socket_accept(listener) + self.assertEqual(client.getsockname(), accepted_address) + self.assertEqual(accepted_address, kernel.socket_peer_address(accepted)) + self.assertEqual(local_address, kernel.socket_local_address(accepted)) + + client.sendall(b"request") + self.assertEqual(b"request", kernel.socket_recv(accepted, 7)) + self.assertEqual(8, kernel.socket_send(accepted, b"response")) + self.assertEqual(b"response", client.recv(8)) + finally: + if accepted is not None: + kernel.socket_close(accepted) + kernel.socket_close(listener) + client.close() + + def test_unix_nonblocking_accept_exhaustion_is_readable_retry(self): + kernel = Unix() + record = kernel.resolve_passive_address("127.0.0.1", 0) + listener = kernel.socket_open(record) + try: + try: + kernel.socket_bind(listener, record) + except PermissionError as exc: + self.skipTest("loopback bind is not permitted: {}".format(exc)) + kernel.socket_listen(listener, 1) + kernel.socket_setblocking(listener, False) + with self.assertRaises(BlockingIOError) as raised: + kernel.socket_accept(listener) + self.assertTrue(kernel.socket_needs_read(raised.exception)) + finally: + kernel.socket_close(listener) + + def test_unix_rejects_passive_record_from_another_kernel(self): + first = Unix() + second = Unix() + record = first.resolve_passive_address("127.0.0.1", 0) + + with self.assertRaises(ValueError): + second.socket_open(record) + with self.assertRaises(ValueError): + second.socket_bind(object(), record) + + def test_peer_address_suppresses_only_not_connected(self): + unix = Unix() + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + self.assertIsNone(unix.socket_peer_address(listener)) + finally: + listener.close() + with self.assertRaises(OSError): + unix.socket_peer_address(listener) + + class FakeErrno: + ENOTCONN = 57 + + stream = FakeStream() + socket_module = TwoArgumentSocketModule(stream) + kernel = MicroPythonKernel( + modules={"socket": socket_module, "errno": FakeErrno()} + ) + stream.peer_error = OSError(57, "not connected") + self.assertIsNone(kernel.socket_peer_address(stream)) + stream.peer_error = OSError(9, "bad descriptor") + with self.assertRaises(OSError): + kernel.socket_peer_address(stream) + + def test_micropython_complete_port_uses_opaque_record_and_constants(self): + stream = FakeStream() + socket_module = TwoArgumentSocketModule(stream) + socket_module.SOL_SOCKET = 7 + socket_module.SO_REUSEADDR = 9 + kernel = MicroPythonKernel(modules={"socket": socket_module}) + + self.assertTrue(kernel.supports_tcp_server()) + self.assertTrue(kernel.supports_reuse_address()) + record = kernel.resolve_passive_address("", 8080) + self.assertNotIsInstance(record, tuple) + self.assertEqual(2, len(socket_module.resolve_calls)) + self.assertEqual(("0.0.0.0", 8080), socket_module.resolve_calls[-1]) + self.assertIs(stream, kernel.socket_open(record)) + + kernel.socket_set_reuse_address(stream, True) + kernel.socket_bind(stream, record) + kernel.socket_listen(stream, 7) + self.assertEqual([(7, 9, 1)], stream.option_calls) + self.assertEqual([("0.0.0.0", 8080)], stream.bind_calls) + self.assertEqual([7], stream.listen_calls) + self.assertEqual(stream.accept_result, kernel.socket_accept(stream)) + self.assertEqual(stream.local_address, kernel.socket_local_address(stream)) + self.assertEqual(stream.peer_address, kernel.socket_peer_address(stream)) + + def test_micropython_does_not_guess_missing_reuse_constants(self): + kernel = MicroPythonKernel( + modules={"socket": TwoArgumentSocketModule(FakeStream())} + ) + + self.assertTrue(kernel.supports_tcp_server()) + self.assertFalse(kernel.supports_reuse_address()) + with self.assertRaises(NotImplementedError): + kernel.socket_set_reuse_address(FakeStream(), True) + + def test_micropython_incomplete_port_reports_unsupported(self): + kernel = MicroPythonKernel(modules={"socket": MissingServerSocketModule()}) + + self.assertFalse(kernel.supports_tcp_server()) + with self.assertRaises(NotImplementedError): + kernel.resolve_passive_address("", 80) + + def test_micropython_incomplete_opened_stream_closes_once(self): + stream = IncompleteStream() + socket_module = TwoArgumentSocketModule(stream) + kernel = MicroPythonKernel(modules={"socket": socket_module}) + + with self.assertRaises(NotImplementedError): + kernel.socket_open(kernel.resolve_passive_address("", 80)) + self.assertEqual(1, stream.close_calls) + + def test_listener_uses_same_record_for_open_and_bind(self): + kernel = RecordingServerKernel() + + listener = _open_listener(kernel, "127.0.0.1", 0, 5) + + self.assertIs(kernel.listener, listener) + self.assertIs(kernel.record, kernel.open_record) + self.assertIs(kernel.record, kernel.bind_args[1]) + self.assertEqual((kernel.listener, False), kernel.blocking_args) + + def test_listener_capability_failure_precedes_resolution_and_binding(self): + kernel = RecordingServerKernel(supported=False) + + with self.assertRaises(NotImplementedError): + _open_listener(kernel, "", 80, 1) + + self.assertEqual(["supports_tcp_server"], kernel.calls) + + def test_listener_post_open_failures_close_once_and_preserve_primary_error(self): + setup_steps = ( + "socket_set_reuse_address", + "socket_bind", + "socket_listen", + "socket_setblocking", + ) + for step in setup_steps: + with self.subTest(step=step): + kernel = RecordingServerKernel(fail_at=step, close_fails=True) + with self.assertRaisesRegex(RuntimeError, "{} failed".format(step)): + _open_listener(kernel, "", 80, 1) + self.assertEqual(1, kernel.calls.count("socket_close")) + self.assertIs(kernel.listener, kernel.closed_listener) + + def test_accepted_stream_failures_close_once_and_preserve_primary_error(self): + for fail_at, spawn_fails, message in ( + ("socket_setblocking", False, "socket_setblocking failed"), + (None, True, "spawn failed"), + ): + with self.subTest(message=message): + kernel = RecordingServerKernel( + fail_at=fail_at, + close_fails=True, + ) + task = DispatchTask(kernel, spawn_fails=spawn_fails) + stream = object() + with self.assertRaisesRegex(RuntimeError, message): + _dispatch_client(task, stream, ("192.0.2.1", 5000), {}) + self.assertEqual(1, kernel.calls.count("socket_close")) + self.assertIs(stream, kernel.closed_listener) + + def test_server_startup_failure_closes_listener_once(self): + kernel = RecordingServerKernel(close_fails=True) + task = ServerTask(kernel, output_fails=True) + + with self.assertRaisesRegex(RuntimeError, "output failed"): + asyncio.run(web_server_task(task, {})) + + self.assertEqual(1, kernel.calls.count("socket_close")) + + def test_server_loop_error_is_not_masked_by_close_failure(self): + kernel = RecordingServerKernel(fail_at="socket_accept", close_fails=True) + task = ServerTask(kernel) + + with self.assertRaisesRegex(RuntimeError, "socket_accept failed"): + asyncio.run(web_server_task(task, {})) + + self.assertEqual(1, kernel.calls.count("socket_close")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/typing/passive_tcp_kernel_contract.py b/tests/typing/passive_tcp_kernel_contract.py new file mode 100644 index 0000000..ad31a3c --- /dev/null +++ b/tests/typing/passive_tcp_kernel_contract.py @@ -0,0 +1,10 @@ +"""Static consumer contract for passive TCP kernels.""" + +from SmallPackage.Kernel import Unix +from SmallPackage._types import PassiveTCPKernelLike + + +kernel: PassiveTCPKernelLike = Unix() +address = kernel.resolve_passive_address("127.0.0.1", 0) +listener = kernel.socket_open(address) +kernel.socket_bind(listener, address)