diff --git a/.coverage b/.coverage deleted file mode 100644 index f1edc58..0000000 Binary files a/.coverage and /dev/null differ diff --git a/.github/workflows/release-wheel.yml b/.github/workflows/release-wheel.yml new file mode 100644 index 0000000..f92ad70 --- /dev/null +++ b/.github/workflows/release-wheel.yml @@ -0,0 +1,35 @@ +name: Release Wheel + +on: + release: + types: [published] + +jobs: + build-wheel: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.8", "3.9", "3.10"] + permissions: + contents: write + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Build wheel + run: | + python -m pip install --upgrade pip + python -m pip install wheel + make wheel + + - name: Upload wheel to release + uses: softprops/action-gh-release@v1 + with: + files: dist/*.whl diff --git a/.gitignore b/.gitignore index 37ebf18..0b697f1 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,7 @@ .pydevproject .settings *.sw* +.coverage +build/ +dist/ +jsocket.egg-info/ diff --git a/Makefile b/Makefile index 86588f1..1fc1ca3 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,14 @@ -.PHONY: test-behave test-pytest-cov test-behave-cov coverage lint +.PHONY: help wheel test-behave test-pytest-cov test-behave-cov coverage lint + +help: + @echo "Targets:" + @echo " help Show this help message" + @echo " wheel Build a wheel into dist/" + @echo " test-behave Run behave tests" + @echo " test-pytest-cov Run pytest with coverage (terminal report)" + @echo " test-behave-cov Run behave with coverage (appends to .coverage)" + @echo " coverage Run combined pytest + behave coverage and export reports" + @echo " lint Run pylint with fail-under threshold" test-behave: PYTHONPATH=. behave -f progress2 @@ -20,7 +30,10 @@ coverage: coverage xml -o coverage.xml coverage html -d .coverage_html -# Static analysis with pylint; fail if score below threshold +# Static analysis with pylint; fail if score below threshold (duplicate-code is noisy in tests/headers) lint: mkdir -p .pylint.d - PYLINTHOME=.pylint.d pylint jsocket tests features/steps --fail-under=9.0 --persistent=n + PYLINTHOME=.pylint.d pylint jsocket tests features/steps --fail-under=9.0 --persistent=n --disable=duplicate-code + +wheel: + python setup.py bdist_wheel diff --git a/features/steps/__init__.py b/features/steps/__init__.py index fdfae28..a23f809 100644 --- a/features/steps/__init__.py +++ b/features/steps/__init__.py @@ -1,2 +1 @@ # Behave step package marker - diff --git a/features/steps/steps.py b/features/steps/steps.py index 8245fb7..bed5321 100644 --- a/features/steps/steps.py +++ b/features/steps/steps.py @@ -1,9 +1,9 @@ +# pylint: disable=not-callable, missing-function-docstring, duplicate-code """Behave step implementations for json socket scenarios. -Note: Pylint is unaware of Behave's decorator callables, so we disable the -"not-callable" check for this file. +Note: Pylint is unaware of Behave's decorator callables, and step helpers +intentionally mirror test setup patterns, so we disable those checks here. """ -# pylint: disable=not-callable, missing-function-docstring import json import logging @@ -90,19 +90,19 @@ def stop_server(context): # Give the thread a moment to terminate cleanly try: server.join(timeout=2.0) - except Exception: + except RuntimeError: pass @then(r"the server is stopped") def see_stopped_server(context): server = getattr(context, 'jsonserver', None) - assert server is not None, "%s" % False + assert server is not None, "server not initialized" # Wait briefly for the thread to terminate deadline = time.time() + 2.0 while server.is_alive() and time.time() < deadline: time.sleep(0.05) - assert not server.is_alive(), "%s" % False + assert not server.is_alive(), "server did not stop in time" @then(r"I close the client") @@ -111,7 +111,7 @@ def close_client(context): if client is not None: try: client.close() - except Exception: + except OSError: pass @@ -139,7 +139,7 @@ def client_attempts_read_with_timeout(context, seconds): context.client_read_value = None try: context.client_read_value = client.read_obj() - except Exception as e: + except Exception as e: # pylint: disable=broad-exception-caught context.client_read_exception = e diff --git a/jsocket/jsocket_base.py b/jsocket/jsocket_base.py index 50016c8..788d5f0 100644 --- a/jsocket/jsocket_base.py +++ b/jsocket/jsocket_base.py @@ -30,6 +30,13 @@ logger = logging.getLogger("jsocket") +def _socket_fileno(sock): + try: + return sock.fileno() + except Exception: # pylint: disable=broad-exception-caught + return None + + class JsonSocket: """Lightweight JSON-over-TCP socket wrapper with length-prefixed framing.""" @@ -85,13 +92,17 @@ def read_obj(self): def close(self): """Close active connection and the listening socket if open.""" - logger.debug("closing all connections") + logger.debug( + "closing sockets (socket fd=%s, conn fd=%s)", + _socket_fileno(self.socket), + _socket_fileno(self.conn), + ) self._close_connection() self._close_socket() def _close_socket(self): """Best-effort shutdown and close of the main socket.""" - logger.debug("closing main socket") + logger.debug("closing main socket (fd=%s)", _socket_fileno(self.socket)) try: if self.socket and self.socket.fileno() != -1: try: @@ -107,7 +118,7 @@ def _close_socket(self): def _close_connection(self): """Best-effort shutdown and close of the accepted connection socket.""" - logger.debug("closing the connection socket") + logger.debug("closing connection socket (fd=%s)", _socket_fileno(self.conn)) try: if self.conn and self.conn is not self.socket and self.conn.fileno() != -1: try: @@ -134,17 +145,17 @@ def _get_address(self): """Return the configured bind address.""" return self._address - def _set_address(self, address): + def _set_address(self, _address): """No-op: address is read-only after initialization.""" - pass + return None def _get_port(self): """Return the configured bind port.""" return self._port - def _set_port(self, port): + def _set_port(self, _port): """No-op: port is read-only after initialization.""" - pass + return None timeout = property(_get_timeout, _set_timeout, doc='Get/set the socket timeout') address = property(_get_address, _set_address, doc='read only property socket address') @@ -159,6 +170,7 @@ def __init__(self, address='127.0.0.1', port=5489): self._bind() def _bind(self): + self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.socket.bind((self.address, self.port)) def _listen(self): @@ -197,11 +209,18 @@ def __init__(self, address='127.0.0.1', port=5489): def connect(self): """Attempt to connect to the server up to 10 times with backoff.""" - for _ in range(10): + for attempt in range(1, 11): try: + logger.debug("connect attempt %d to %s:%s", attempt, self.address, self.port) self.socket.connect((self.address, self.port)) except socket.error as msg: logger.error("SockThread Error: %s", msg) + # Recreate the socket to avoid retrying on a potentially bad fd. + self._close_socket() + self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.socket.settimeout(self._timeout) + self.conn = self.socket + logger.debug("recreated socket for retry %d to %s:%s", attempt, self.address, self.port) time.sleep(3) continue logger.info("...Socket Connected") diff --git a/jsocket/tserver.py b/jsocket/tserver.py index 82a6cdc..a9bb173 100644 --- a/jsocket/tserver.py +++ b/jsocket/tserver.py @@ -33,6 +33,14 @@ logger = logging.getLogger("jsocket.tserver") +def _response_summary(resp_obj) -> str: + if isinstance(resp_obj, dict): + return f"type=dict keys={len(resp_obj)}" + if isinstance(resp_obj, (list, tuple)): + return f"type={type(resp_obj).__name__} items={len(resp_obj)}" + return f"type={type(resp_obj).__name__}" + + class ThreadedServer(threading.Thread, jsocket_base.JsonServer, metaclass=abc.ABCMeta): """Single-threaded server that accepts one connection and processes messages in its thread.""" @@ -53,45 +61,54 @@ def _process_message(self, obj) -> Optional[dict]: # Return None in the base class to satisfy linters; subclasses should override. return None - def run(self): - # Ensure the run loop is active even when run() is invoked directly - # (tests may call run() in a separate thread without invoking start()). - if not self._is_alive: - self._is_alive = True + def _accept_client(self) -> bool: + """Accept an incoming connection; return True when a client connects.""" + try: + self.accept_connection() + except socket.timeout as e: + logger.debug("accept timeout on %s:%s: %s", self.address, self.port, e) + return False + except Exception as e: # pylint: disable=broad-exception-caught + # Avoid noisy error logs during normal shutdown/sequencing + if self._is_alive: + logger.debug("accept error on %s:%s: %s", self.address, self.port, e) + return False + logger.debug("server stopping; accept loop exiting (%s:%s)", self.address, self.port) + self._is_alive = False + return False + return True + + def _handle_client_messages(self): + """Read, process, and respond to client messages until disconnect.""" while self._is_alive: try: - self.accept_connection() + obj = self.read_obj() + resp_obj = self._process_message(obj) + if resp_obj is not None: + logger.debug("sending response (%s)", _response_summary(resp_obj)) + self.send_obj(resp_obj) except socket.timeout as e: - logger.debug("socket.timeout: %s", e) + logger.debug("read timeout waiting for client data: %s", e) continue except Exception as e: # pylint: disable=broad-exception-caught - # Avoid noisy error logs during normal shutdown/sequencing - if self._is_alive: - logger.debug("accept_connection error: %s", e) + # Treat client disconnects as normal; keep logs at info/debug + msg = str(e) + if isinstance(e, RuntimeError) and 'socket connection broken' in msg: + logger.info("client connection broken, closing connection") else: - logger.debug("server stopping; accept loop exiting") - break - continue + logger.debug("handler error (%s): %s", type(e).__name__, e) + self._close_connection() + break - while self._is_alive: - try: - obj = self.read_obj() - resp_obj = self._process_message(obj) - if resp_obj is not None: - logger.debug("message has a response") - self.send_obj(resp_obj) - except socket.timeout as e: - logger.debug("socket.timeout: %s", e) - continue - except Exception as e: # pylint: disable=broad-exception-caught - # Treat client disconnects as normal; keep logs at info/debug - msg = str(e) - if isinstance(e, RuntimeError) and 'socket connection broken' in msg: - logger.info("client connection broken, closing connection") - else: - logger.debug("handler error: %s", e) - self._close_connection() - break + def run(self): + # Ensure the run loop is active even when run() is invoked directly + # (tests may call run() in a separate thread without invoking start()). + if not self._is_alive: + self._is_alive = True + while self._is_alive: + if not self._accept_client(): + continue + self._handle_client_messages() # Ensure sockets are cleaned up when the server stops try: self.close() @@ -106,7 +123,7 @@ def start(self): """ self._is_alive = True super().start() - logger.debug("Threaded Server has been started.") + logger.debug("Threaded Server started on %s:%s", self.address, self.port) def stop(self): """ Stops the threaded server. @@ -115,7 +132,7 @@ def stop(self): @retval None """ self._is_alive = False - logger.debug("Threaded Server has been stopped.") + logger.debug("Threaded Server stopped on %s:%s", self.address, self.port) class ServerFactoryThread(threading.Thread, jsocket_base.JsonSocket, metaclass=abc.ABCMeta): @@ -123,6 +140,8 @@ class ServerFactoryThread(threading.Thread, jsocket_base.JsonSocket, metaclass=a def __init__(self, **kwargs): threading.Thread.__init__(self, **kwargs) + self.socket = None + self.conn = None jsocket_base.JsonSocket.__init__(self, **kwargs) self._is_alive = False @@ -144,16 +163,18 @@ def run(self): obj = self.read_obj() resp_obj = self._process_message(obj) if resp_obj is not None: - logger.debug("message has a response") + logger.debug("sending response (%s)", _response_summary(resp_obj)) self.send_obj(resp_obj) except socket.timeout as e: - logger.debug("socket.timeout: %s", e) + logger.debug("worker read timeout waiting for data: %s", e) continue except Exception as e: # pylint: disable=broad-exception-caught logger.info("client connection broken, closing connection: %s", e) self._is_alive = False break self._close_connection() + if hasattr(self, "socket"): + self._close_socket() @abc.abstractmethod def _process_message(self, obj) -> Optional[dict]: @@ -173,7 +194,7 @@ def start(self): """ self._is_alive = True super().start() - logger.debug("ServerFactoryThread has been started.") + logger.debug("ServerFactoryThread started (%s)", self.name) def force_stop(self): """ Force stops the factory thread. @@ -183,7 +204,7 @@ def force_stop(self): @retval None """ self._is_alive = False - logger.debug("ServerFactoryThread has been stopped.") + logger.debug("ServerFactoryThread stopped (%s)", self.name) class ServerFactory(ThreadedServer): @@ -214,7 +235,7 @@ def run(self): try: self.accept_connection() except socket.timeout as e: - logger.debug("socket.timeout: %s", e) + logger.debug("factory accept timeout on %s:%s: %s", self.address, self.port, e) continue except Exception as e: # pylint: disable=broad-exception-caught logger.exception("accept error: %s", e) @@ -233,6 +254,7 @@ def run(self): self.close() def stop_all(self): + """Stop and join all active worker threads.""" for t in self._threads: if t.is_alive(): t.force_stop() @@ -249,9 +271,10 @@ def stop(self): self.stop_all() except Exception: # pylint: disable=broad-exception-caught pass - logger.debug("ServerFactory has been stopped.") + logger.debug("ServerFactory stopped on %s:%s", self.address, self.port) def _wait_to_exit(self): + """Block until all worker threads have finished.""" while self._get_num_of_active_threads(): time.sleep(0.2) diff --git a/tests/test_additional_coverage.py b/tests/test_additional_coverage.py index 9bdda6d..d4fa5b5 100644 --- a/tests/test_additional_coverage.py +++ b/tests/test_additional_coverage.py @@ -48,31 +48,57 @@ def test_close_idempotent_and_connected_guard(): s.close() -def test_threadedserver_timeout_then_exception_triggers_close(monkeypatch): +def test_jsonserver_sets_reuseaddr(monkeypatch): + """JsonServer should set SO_REUSEADDR before binding.""" + calls = [] + orig_setsockopt = socket.socket.setsockopt + + def tracking_setsockopt(self, level, optname, value): + calls.append((level, optname, value)) + return orig_setsockopt(self, level, optname, value) + + monkeypatch.setattr(socket.socket, "setsockopt", tracking_setsockopt, raising=True) + + try: + server = jsocket.JsonServer(address="127.0.0.1", port=0) + except PermissionError as e: # sandboxed environments may forbid sockets + pytest.skip(f"Socket creation blocked: {e}") + server.close() + + assert any( + level == socket.SOL_SOCKET and optname == socket.SO_REUSEADDR and value in (1, True) + for level, optname, value in calls + ) + + +def test_threadedserver_timeout_then_exception_triggers_close(): """ThreadedServer should ignore timeouts and close on generic exceptions.""" class ProbeServer(jsocket.ThreadedServer): - def __init__(self): + """ThreadedServer stub that avoids real sockets.""" + # pylint: disable=non-parent-init-called + + def __init__(self): # pylint: disable=super-init-not-called # Do not call super to avoid binding sockets threading.Thread.__init__(self) self._is_alive = True - self._close_calls = 0 - self._reads = iter([ - lambda: (_ for _ in ()).throw(socket.timeout("t")), - lambda: (_ for _ in ()).throw(ValueError("boom")), - ]) + self._address = "127.0.0.1" + self._port = 0 + self._state = 0 + self.close_calls = 0 def accept_connection(self): + """No-op; simulate an accepted connection.""" # No-op; simulate an accepted connection return None def read_obj(self): + """Simulate timeout, then error, then timeouts.""" # First a timeout, then an exception, then timeouts until stopped - st = getattr(self, "_state", 0) - if st == 0: + if self._state == 0: self._state = 1 raise socket.timeout("t") - if st == 1: + if self._state == 1: self._state = 2 raise ValueError("boom") raise socket.timeout("t") @@ -81,9 +107,10 @@ def _process_message(self, obj): # pragma: no cover - not reached return None def _close_connection(self): - self._close_calls += 1 + self.close_calls += 1 def close(self): + """No-op: avoid touching real sockets.""" # Avoid base close touching real sockets return None @@ -94,14 +121,17 @@ def close(self): srv.stop() srv.join(timeout=1.0) # One close due to ValueError path - assert srv._close_calls == 1 + assert srv.close_calls == 1 def test_serverfactorythread_exception_closes_connection(): """ServerFactoryThread should close the connection when handler raises.""" class BoomWorker(jsocket.ServerFactoryThread): - def __init__(self): + """Worker that raises to trigger close handling.""" + # pylint: disable=non-parent-init-called + + def __init__(self): # pylint: disable=super-init-not-called # Avoid base JsonSocket init threading.Thread.__init__(self) self._is_alive = True @@ -111,6 +141,7 @@ def _process_message(self, obj): # pylint: disable=unused-argument raise ValueError("boom") def read_obj(self): + """Return a single payload to trigger processing.""" return {"echo": 1} def _close_connection(self): @@ -122,11 +153,53 @@ def _close_connection(self): assert w.closed is True +def test_serverfactorythread_closes_socket_after_run(): + """ServerFactoryThread should close the swapped socket when exiting.""" + if not hasattr(socket, "socketpair"): + pytest.skip("socketpair unavailable") + + try: + client_sock, server_sock = socket.socketpair() + except OSError as e: + pytest.skip(f"Socketpair blocked: {e}") + + class CloseWorker(jsocket.ServerFactoryThread): + """Worker that reuses an existing socket.""" + # pylint: disable=non-parent-init-called + + def __init__(self, sock): # pylint: disable=super-init-not-called + threading.Thread.__init__(self) + self._is_alive = False + self.socket = sock + self.conn = sock + + def _process_message(self, obj): # pragma: no cover - not used here + return None + + worker = CloseWorker(server_sock) + try: + assert worker.socket.fileno() != -1 + worker.run() + assert worker.socket.fileno() == -1 + finally: + try: + client_sock.close() + except OSError: + pass + try: + server_sock.close() + except OSError: + pass + + def test_serverfactory_accept_error_branch(monkeypatch): """ServerFactory should continue on accept() errors and then stop cleanly.""" class EchoWorker(jsocket.ServerFactoryThread): - def __init__(self): + """Worker stub for accept error branch test.""" + # pylint: disable=non-parent-init-called + + def __init__(self): # pylint: disable=super-init-not-called threading.Thread.__init__(self) self._is_alive = False @@ -145,8 +218,9 @@ def flappy_accept(): calls["n"] += 1 if calls["n"] == 1: raise RuntimeError("accept failed") - # On second call, request stop - server._is_alive = False + # On second call, request stop without accepting a connection + server.stop() + raise RuntimeError("accept stopped") monkeypatch.setattr(server, "accept_connection", flappy_accept) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 85b6f34..f7e6855 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -31,6 +31,7 @@ def test_end_to_end_echo_and_connection(): except PermissionError as e: pytest.skip(f"Socket creation blocked: {e}") + client = None # Discover the ephemeral port chosen by the OS _, port = server.socket.getsockname() server.start() @@ -54,9 +55,10 @@ def test_end_to_end_echo_and_connection(): assert echoed == payload finally: # Cleanup - try: - client.close() - except Exception: - pass + if client is not None: + try: + client.close() + except OSError: + pass server.stop() server.join(timeout=3) diff --git a/tests/test_serverfactory_concurrent.py b/tests/test_serverfactory_concurrent.py index 057381d..26aac82 100644 --- a/tests/test_serverfactory_concurrent.py +++ b/tests/test_serverfactory_concurrent.py @@ -56,10 +56,10 @@ def test_serverfactory_handles_two_clients_concurrently(): c1.close() c2.close() finally: - try: - if hasattr(server, 'stop_all'): + if hasattr(server, 'stop_all'): + try: server.stop_all() - except Exception: - pass + except RuntimeError: + pass server.stop() server.join(timeout=3) diff --git a/tests/test_serverfactory_serialization.py b/tests/test_serverfactory_serialization.py index adc14f8..a72894b 100644 --- a/tests/test_serverfactory_serialization.py +++ b/tests/test_serverfactory_serialization.py @@ -65,11 +65,11 @@ def test_serverfactory_accepts_multiple_active_clients_concurrently(): c1.close() finally: # Stop the server thread and join - try: - # Ensure any worker threads are stopped to avoid hangs - if hasattr(server, 'stop_all'): + # Ensure any worker threads are stopped to avoid hangs + if hasattr(server, 'stop_all'): + try: server.stop_all() - except Exception: - pass + except RuntimeError: + pass server.stop() server.join(timeout=3)