diff --git a/Makefile b/Makefile index f7b75f1..21e6328 100644 --- a/Makefile +++ b/Makefile @@ -1,20 +1,38 @@ -.PHONY: help wheel test-behave test test-behave-cov coverage lint publish version +.PHONY: help clean wheel test-behave test test-behave-cov coverage lint publish version net-server net-client help: @echo "Targets:" @echo " help Show this help message" + @echo " clean Remove build/test artifacts" @echo " wheel Build a wheel into dist/" @echo " test-behave Run behave tests" @echo " test 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" + @echo " net-server Run echo server for network testing (IP required)" + @echo " net-client Run client for network testing (IP required)" @echo " publish Upload dist/* to PyPI via twine" @echo " version Show package version and git SHA" test-behave: PYTHONPATH=. behave -f progress2 +PORT ?= 5491 +MODE ?= ping +NUM ?= 1 +MAX ?= 100 +ERROR ?= 0 +READ ?= 1 + +net-server: + @if [ -z "$(IP)" ]; then echo "Usage: make net-server IP= [PORT=$(PORT)]"; exit 1; fi + PYTHONPATH=. python3 scripts/net_server.py $(IP) --port $(PORT) + +net-client: + @if [ -z "$(IP)" ]; then echo "Usage: make net-client IP= [PORT=$(PORT)] [MODE=$(MODE)] [NUM=$(NUM)] [MAX=$(MAX)] [ERROR=$(ERROR)] [READ=$(READ)]"; exit 1; fi + PYTHONPATH=. python3 scripts/net_client.py $(IP) --port $(PORT) --mode $(MODE) --num $(NUM) --count $(MAX) --error $(ERROR) --read $(READ) + # Pytest coverage (terminal report) test: pytest -q --cov=jsocket --cov-branch --cov-report=term-missing @@ -37,6 +55,9 @@ lint: mkdir -p .pylint.d PYLINTHOME=.pylint.d pylint jsocket tests features/steps --fail-under=9.0 --persistent=n --disable=duplicate-code +clean: + rm -rf build dist *.egg-info .pytest_cache .coverage coverage.xml .coverage_html + wheel: python3 -m build --wheel diff --git a/jsocket/jsocket_base.py b/jsocket/jsocket_base.py index 4131dcc..be9c345 100644 --- a/jsocket/jsocket_base.py +++ b/jsocket/jsocket_base.py @@ -72,6 +72,8 @@ def __init__( self._last_client_addr = None self._is_server = False self._is_listening = False + self._last_read_size = None + self._last_send_size = None if create_socket: self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.conn = self.socket @@ -83,6 +85,7 @@ def send_obj(self, obj): msg = json.dumps(obj, ensure_ascii=False) if self.socket: payload = msg.encode('utf-8') + self._last_send_size = len(payload) if self._max_message_size is not None and len(payload) > self._max_message_size: raise ValueError(f"message exceeds max_message_size ({len(payload)} > {self._max_message_size})") checksum = zlib.crc32(payload) & 0xFFFFFFFF @@ -136,6 +139,7 @@ def _read_header(self): def read_obj(self): """Read a full message and decode it as JSON, returning a Python object.""" size, checksum = self._read_header() + self._last_read_size = size data = self._read(size) actual = zlib.crc32(data) & 0xFFFFFFFF if actual != checksum: diff --git a/jsocket/tserver.py b/jsocket/tserver.py index 326fbb2..99c82a6 100644 --- a/jsocket/tserver.py +++ b/jsocket/tserver.py @@ -27,6 +27,7 @@ import logging import abc from typing import Optional +from contextlib import contextmanager from jsocket import jsocket_base from ._version import __version__ @@ -52,6 +53,303 @@ def _format_client_id(addr) -> str: return "unknown" +def _now_ts() -> float: + return time.time() + + +def _new_failure_counts() -> dict: + return { + "timeout": 0, + "bad_write": 0, + "bad_crc": 0, + "bad_header": 0, + "oversize": 0, + "invalid_utf8": 0, + "invalid_json": 0, + "handler": 0, + "framing": 0, + } + + +def _new_client_stats(client_id: str) -> dict: + return { + "client_id": client_id, + "connected": False, + "connects": 0, + "disconnects": 0, + "messages_in": 0, + "messages_out": 0, + "bytes_in": 0, + "bytes_out": 0, + "total_connected_duration": 0.0, + "failures": _new_failure_counts(), + "last_connect_ts": None, + "last_disconnect_ts": None, + "last_message_ts": None, + "_connected_since": None, + } + + +def _clone_client_stats(stats: dict) -> dict: + return { + **stats, + "failures": dict(stats.get("failures", {})), + } + + +def _format_client_stats(stats: dict, now_mono: float) -> dict: + snapshot = _clone_client_stats(stats) + messages_in = snapshot.get("messages_in", 0) or 0 + messages_out = snapshot.get("messages_out", 0) or 0 + bytes_in = snapshot.get("bytes_in", 0) or 0 + bytes_out = snapshot.get("bytes_out", 0) or 0 + snapshot["avg_payload_in"] = bytes_in / messages_in if messages_in else 0.0 + snapshot["avg_payload_out"] = bytes_out / messages_out if messages_out else 0.0 + connected = snapshot.get("connected") is True + connected_since = snapshot.get("_connected_since") + current_duration = now_mono - connected_since if (connected and connected_since) else 0.0 + snapshot["connected_duration"] = current_duration + total = snapshot.get("total_connected_duration", 0.0) or 0.0 + snapshot["total_connected_duration"] = total + current_duration if connected else total + snapshot.pop("_connected_since", None) + return snapshot + + +@contextmanager +def _stats_guard(obj): + lock = getattr(obj, "_stats_lock", None) + if lock is None: + yield + else: + with lock: + yield + + +def _ensure_stats_state(obj): + if not hasattr(obj, "_client_stats"): + obj._client_stats = {} + if not hasattr(obj, "_active_client_id"): + obj._active_client_id = None + + +def _get_active_client_id(obj): + client_id = getattr(obj, "_active_client_id", None) + if client_id: + return client_id + return getattr(obj, "_client_id", None) + + +def _get_or_create_stats(obj, client_id: str) -> dict: + _ensure_stats_state(obj) + stats = obj._client_stats.get(client_id) + if stats is None: + stats = _new_client_stats(client_id) + obj._client_stats[client_id] = stats + return stats + + +def _note_connect(obj, client_id: str) -> None: + if not client_id: + client_id = "unknown" + with _stats_guard(obj): + stats = _get_or_create_stats(obj, client_id) + stats["connected"] = True + stats["connects"] += 1 + stats["last_connect_ts"] = _now_ts() + if stats.get("_connected_since") is None: + stats["_connected_since"] = time.monotonic() + obj._active_client_id = client_id + + +def _note_disconnect(obj) -> None: + client_id = _get_active_client_id(obj) + if not client_id: + return + with _stats_guard(obj): + stats = _get_or_create_stats(obj, client_id) + if not stats.get("connected"): + return + stats["connected"] = False + stats["disconnects"] += 1 + stats["last_disconnect_ts"] = _now_ts() + connected_since = stats.get("_connected_since") + if connected_since is not None: + duration = time.monotonic() - connected_since + stats["total_connected_duration"] = (stats.get("total_connected_duration") or 0.0) + duration + stats["_connected_since"] = None + obj._active_client_id = None + + +def _note_message_in(obj, size) -> None: + client_id = _get_active_client_id(obj) + if not client_id: + return + msg_size = int(size) if size is not None else 0 + with _stats_guard(obj): + stats = _get_or_create_stats(obj, client_id) + stats["messages_in"] += 1 + stats["bytes_in"] += msg_size + stats["last_message_ts"] = _now_ts() + + +def _note_message_out(obj, size) -> None: + client_id = _get_active_client_id(obj) + if not client_id: + return + msg_size = int(size) if size is not None else 0 + with _stats_guard(obj): + stats = _get_or_create_stats(obj, client_id) + stats["messages_out"] += 1 + stats["bytes_out"] += msg_size + stats["last_message_ts"] = _now_ts() + + +def _note_failure(obj, kind: str) -> None: + client_id = _get_active_client_id(obj) + if not client_id: + return + with _stats_guard(obj): + stats = _get_or_create_stats(obj, client_id) + failures = stats.get("failures") + if failures is None: + failures = _new_failure_counts() + stats["failures"] = failures + failures[kind] = failures.get(kind, 0) + 1 + + +def _framing_failure_kind(error: Exception) -> str: + msg = str(error) + if "invalid message header magic" in msg: + return "bad_header" + if "checksum mismatch" in msg: + return "bad_crc" + if "exceeds max_message_size" in msg: + return "oversize" + if "invalid UTF-8 payload" in msg: + return "invalid_utf8" + if "invalid JSON payload" in msg: + return "invalid_json" + if "socket read timeout" in msg: + return "timeout" + return "framing" + + +def _note_framing_failure(obj, error: Exception) -> None: + _note_failure(obj, _framing_failure_kind(error)) + + +def _max_ts(left, right): + if left is None: + return right + if right is None: + return left + return right if right > left else left + + +def _normalize_client_id(value): + if value is None: + return None + if isinstance(value, (int, float)): + return str(value) + if isinstance(value, str): + return value.strip() or None + return str(value) + + +def _extract_client_id(obj): + if not isinstance(obj, dict): + return None + if "client" in obj: + return _normalize_client_id(obj.get("client")) + if "client_id" in obj: + return _normalize_client_id(obj.get("client_id")) + return None + + +def _set_client_identity(obj, new_client_id: str) -> None: + if not new_client_id: + return + current_id = _get_active_client_id(obj) + if current_id == new_client_id: + return + _ensure_stats_state(obj) + with _stats_guard(obj): + existing = obj._client_stats.get(current_id) if current_id else None + if existing is None: + existing = _new_client_stats(new_client_id) + existing["client_id"] = new_client_id + target = obj._client_stats.get(new_client_id) + if target is None: + obj._client_stats[new_client_id] = existing + else: + obj._client_stats[new_client_id] = _merge_client_stats(target, existing) + if current_id and current_id != new_client_id: + obj._client_stats.pop(current_id, None) + obj._active_client_id = new_client_id + if hasattr(obj, "_client_id"): + obj._client_id = new_client_id + + +def _merge_client_stats(dest: dict, src: dict) -> dict: + if not dest: + return _clone_client_stats(src) + dest.setdefault("client_id", src.get("client_id")) + for key in ("connects", "disconnects", "messages_in", "messages_out", "bytes_in", "bytes_out"): + dest[key] = dest.get(key, 0) + (src.get(key, 0) or 0) + dest["total_connected_duration"] = (dest.get("total_connected_duration") or 0.0) + ( + src.get("total_connected_duration") or 0.0 + ) + dest_failures = dest.get("failures") + if dest_failures is None: + dest_failures = _new_failure_counts() + dest["failures"] = dest_failures + for key, value in (src.get("failures") or {}).items(): + dest_failures[key] = dest_failures.get(key, 0) + (value or 0) + dest["last_connect_ts"] = _max_ts(dest.get("last_connect_ts"), src.get("last_connect_ts")) + dest["last_disconnect_ts"] = _max_ts(dest.get("last_disconnect_ts"), src.get("last_disconnect_ts")) + dest["last_message_ts"] = _max_ts(dest.get("last_message_ts"), src.get("last_message_ts")) + if src.get("connected"): + dest["connected"] = True + src_since = src.get("_connected_since") + dest_since = dest.get("_connected_since") + if dest_since is None or (src_since is not None and src_since < dest_since): + dest["_connected_since"] = src_since + return dest + + +def _rekey_stats_map(stats_map: dict) -> dict: + if not stats_map: + return {} + rekeyed = {} + for client_id, stats in stats_map.items(): + key = stats.get("client_id") or client_id + if key in rekeyed: + rekeyed[key] = _merge_client_stats(rekeyed[key], stats) + else: + rekeyed[key] = _clone_client_stats(stats) + rekeyed[key]["client_id"] = key + return rekeyed + + +def _stats_from_thread(thread) -> dict: + if hasattr(thread, "_get_client_stats_internal"): + return thread._get_client_stats_internal() + client_id = getattr(thread, "_client_id", None) + if not client_id: + name = getattr(thread, "name", None) + if name: + client_id = f"thread-{name}" + if not client_id: + return {} + stats = _new_client_stats(client_id) + stats["connected"] = True + stats["connects"] = 1 + started_at = getattr(thread, "_client_started_at", None) + if started_at is not None: + stats["_connected_since"] = started_at + return {client_id: stats} + + class ThreadedServer(threading.Thread, jsocket_base.JsonServer, metaclass=abc.ABCMeta): """Single-threaded server that accepts one connection and processes messages in its thread.""" @@ -62,6 +360,8 @@ def __init__(self, **kwargs): self._stats_lock = threading.Lock() self._client_started_at = None self._client_id = None + self._client_stats = {} + self._active_client_id = None self._wakeup_r = None self._wakeup_w = None self._init_wakeup() @@ -182,24 +482,28 @@ def _record_client_start(self): addr = self.conn.getpeername() except OSError: addr = None + client_id = _format_client_id(addr) with self._stats_lock: self._client_started_at = time.monotonic() - self._client_id = _format_client_id(addr) + self._client_id = client_id + _note_connect(self, client_id) def _clear_client_stats(self): with self._stats_lock: self._client_started_at = None self._client_id = None + self._active_client_id = None def get_client_stats(self) -> dict: - """Return connected client count and per-client durations in seconds.""" - with self._stats_lock: - started_at = self._client_started_at - client_id = self._client_id - if not started_at or not client_id or not self.connected: - return {"connected_clients": 0, "clients": {}} - duration = time.monotonic() - started_at - return {"connected_clients": 1, "clients": {client_id: duration}} + """Return per-client stats including connects, messages, failures, and timestamps.""" + _ensure_stats_state(self) + with _stats_guard(self): + stats_map = {cid: _clone_client_stats(stats) for cid, stats in self._client_stats.items()} + stats_map = _rekey_stats_map(stats_map) + now = time.monotonic() + clients = {cid: _format_client_stats(stats, now) for cid, stats in stats_map.items()} + connected = sum(1 for stats in clients.values() if stats.get("connected")) + return {"connected_clients": connected, "clients": clients} def _accept_client(self) -> bool: """Accept an incoming connection; return True when a client connects.""" @@ -226,13 +530,15 @@ def _handle_client_messages(self): while self._is_alive: try: 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("read timeout waiting for client data: %s", e) + _note_failure(self, "timeout") continue + except jsocket_base.FramingError as e: + _note_framing_failure(self, e) + logger.debug("framing error (%s): %s", type(e).__name__, e) + self._close_connection() + break except Exception as e: # pylint: disable=broad-exception-caught # Treat client disconnects as normal; keep logs at info/debug msg = str(e) @@ -240,8 +546,35 @@ def _handle_client_messages(self): logger.info("client connection broken, closing connection") else: logger.debug("handler error (%s): %s", type(e).__name__, e) + _note_failure(self, "handler") self._close_connection() break + client_id = _extract_client_id(obj) + if client_id: + _set_client_identity(self, client_id) + _note_message_in(self, getattr(self, "_last_read_size", None)) + try: + resp_obj = self._process_message(obj) + except Exception as e: # pylint: disable=broad-exception-caught + logger.debug("handler error (%s): %s", type(e).__name__, e) + _note_failure(self, "handler") + self._close_connection() + break + if resp_obj is not None: + logger.debug("sending response (%s)", _response_summary(resp_obj)) + try: + self.send_obj(resp_obj) + except Exception as e: # pylint: disable=broad-exception-caught + msg = str(e) + if isinstance(e, RuntimeError) and 'socket connection broken' in msg: + logger.info("client connection broken, closing connection") + else: + logger.debug("send error (%s): %s", type(e).__name__, e) + _note_failure(self, "bad_write") + self._close_connection() + break + _note_message_out(self, getattr(self, "_last_send_size", None)) + _note_disconnect(self) self._clear_client_stats() def run(self): @@ -277,7 +610,6 @@ def stop(self): @retval None """ self._is_alive = False - self._clear_client_stats() self._signal_wakeup() logger.debug("Threaded Server stopped on %s:%s", self.address, self.port) @@ -296,8 +628,11 @@ def __init__(self, **kwargs): self.conn = None jsocket_base.JsonSocket.__init__(self, create_socket=create_socket, **kwargs) self._is_alive = False + self._stats_lock = threading.Lock() self._client_started_at = None self._client_id = None + self._client_stats = {} + self._active_client_id = None def swap_socket(self, new_sock): """ Swaps the existing socket with a new one. Useful for setting socket after a new connection. @@ -325,6 +660,7 @@ def swap_socket(self, new_sock): addr = None self._client_id = _format_client_id(addr) self._client_started_at = time.monotonic() + _note_connect(self, self._client_id) def run(self): """ Should exit when client closes socket conn. @@ -333,17 +669,50 @@ def run(self): while self._is_alive: try: 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("worker read timeout waiting for data: %s", e) + _note_failure(self, "timeout") continue + except jsocket_base.FramingError as e: + _note_framing_failure(self, e) + logger.debug("worker framing error (%s): %s", type(e).__name__, e) + self._is_alive = False + break except Exception as e: # pylint: disable=broad-exception-caught - logger.info("client connection broken, closing connection: %s", e) + msg = str(e) + if isinstance(e, RuntimeError) and "socket connection broken" in msg: + logger.info("client connection broken, closing connection") + else: + logger.debug("worker error (%s): %s", type(e).__name__, e) + _note_failure(self, "handler") self._is_alive = False break + client_id = _extract_client_id(obj) + if client_id: + _set_client_identity(self, client_id) + _note_message_in(self, getattr(self, "_last_read_size", None)) + try: + resp_obj = self._process_message(obj) + except Exception as e: # pylint: disable=broad-exception-caught + logger.debug("worker handler error (%s): %s", type(e).__name__, e) + _note_failure(self, "handler") + self._is_alive = False + break + if resp_obj is not None: + logger.debug("sending response (%s)", _response_summary(resp_obj)) + try: + self.send_obj(resp_obj) + except Exception as e: # pylint: disable=broad-exception-caught + msg = str(e) + if isinstance(e, RuntimeError) and "socket connection broken" in msg: + logger.info("client connection broken, closing connection") + else: + logger.debug("worker send error (%s): %s", type(e).__name__, e) + _note_failure(self, "bad_write") + self._is_alive = False + break + _note_message_out(self, getattr(self, "_last_send_size", None)) + _note_disconnect(self) self._close_connection() if hasattr(self, "socket"): self._close_socket() @@ -378,6 +747,11 @@ def force_stop(self): self._is_alive = False logger.debug("ServerFactoryThread stopped (%s)", self.name) + def _get_client_stats_internal(self) -> dict: + _ensure_stats_state(self) + with _stats_guard(self): + return {cid: _clone_client_stats(stats) for cid, stats in self._client_stats.items()} + class ServerFactory(ThreadedServer): """Accepts clients and spawns a ServerFactoryThread per connection.""" @@ -398,6 +772,7 @@ def __init__(self, server_thread, **kwargs): self._thread_type = server_thread self._threads = [] self._threads_lock = threading.Lock() + self._client_stats_archive = {} self._thread_args = kwargs self._thread_args.pop('address', None) self._thread_args.pop('port', None) @@ -497,10 +872,42 @@ def stop_all(self): t.join() self._purge_threads() + def _archive_thread_stats(self, thread): + if getattr(thread, "_stats_archived", False): + return + stats_map = _stats_from_thread(thread) + if not stats_map: + thread._stats_archived = True + return + # Dead threads should never be marked connected in the archive. + for stats in stats_map.values(): + stats["connected"] = False + stats["_connected_since"] = None + with _stats_guard(self): + archive = getattr(self, "_client_stats_archive", None) + if archive is None: + self._client_stats_archive = {} + archive = self._client_stats_archive + for client_id, stats in stats_map.items(): + if client_id not in archive: + archive[client_id] = _clone_client_stats(stats) + else: + archive[client_id] = _merge_client_stats(archive[client_id], stats) + thread._stats_archived = True + def _purge_threads(self): - # Rebuild list to avoid mutating while iterating + # Rebuild list to avoid mutating while iterating, archiving stats for finished threads. with self._threads_lock: - self._threads = [t for t in self._threads if t.is_alive()] + alive = [] + dead = [] + for t in self._threads: + if t.is_alive(): + alive.append(t) + else: + dead.append(t) + self._threads = alive + for t in dead: + self._archive_thread_stats(t) def stop(self): # Stop accepting and stop all workers @@ -523,20 +930,29 @@ def _get_num_of_active_threads(self): return len([True for x in threads if x.is_alive()]) def get_client_stats(self) -> dict: - """Return connected client count and per-client durations in seconds.""" + """Return per-client stats including connects, messages, failures, and timestamps.""" with self._threads_lock: threads = list(self._threads) - now = time.monotonic() - clients = {} - active = 0 + # Archive any finished threads encountered. for t in threads: if not t.is_alive(): - continue - active += 1 - started_at = getattr(t, "_client_started_at", None) - client_id = getattr(t, "_client_id", None) or f"thread-{t.name}" - duration = now - started_at if started_at else 0.0 - clients[client_id] = duration - return {"connected_clients": active, "clients": clients} + self._archive_thread_stats(t) + alive = [t for t in threads if t.is_alive()] + with _stats_guard(self): + archive = getattr(self, "_client_stats_archive", {}) or {} + combined = {cid: _clone_client_stats(stats) for cid, stats in archive.items()} + for t in alive: + stats_map = _stats_from_thread(t) + for client_id, stats in stats_map.items(): + existing = combined.get(client_id) + if existing is None: + combined[client_id] = _clone_client_stats(stats) + else: + combined[client_id] = _merge_client_stats(existing, stats) + combined = _rekey_stats_map(combined) + now = time.monotonic() + clients = {cid: _format_client_stats(stats, now) for cid, stats in combined.items()} + connected = sum(1 for stats in clients.values() if stats.get("connected")) + return {"connected_clients": connected, "clients": clients} active = property(_get_num_of_active_threads, doc="number of active threads") diff --git a/scripts/net_client.py b/scripts/net_client.py new file mode 100644 index 0000000..3222ad1 --- /dev/null +++ b/scripts/net_client.py @@ -0,0 +1,367 @@ +import argparse +import json +import logging +import secrets +import socket +import struct +import sys +import threading +import time +import zlib +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import jsocket +from jsocket import jsocket_base + + +logger = logging.getLogger("jsocket.net_client") +logging.basicConfig(level=logging.INFO, format='[%(levelname)s] %(message)s') + +PERF_PAD_RANGES = ( + (0, 0), + (8, 64), + (64, 256), + (256, 1024), + (1024, 4096), + (4096, 16384), + (16384, 65536), + (65536, 131072), +) +ADAPTIVE_SLOW_PAD_RANGE = (0, 256) +ADAPTIVE_BURST_PAD_RANGE = (0, 8192) +ADAPTIVE_LARGE_PAD_RANGE = (65536, 262144) +ERROR_RATE = 0.05 + + +def parse_args(argv): + parser = argparse.ArgumentParser( + description="Minimal JSON socket client for quick network testing.", + ) + parser.add_argument("host", help="Server IP or hostname") + parser.add_argument( + "--port", + type=int, + default=5491, + help="Server port (default: 5491)", + ) + parser.add_argument( + "--message", + help="JSON payload to send (defaults to {\"message\": \"ping\"})", + ) + parser.add_argument( + "--mode", + choices=("ping", "performance", "adaptive"), + default="ping", + help="Client mode (default: ping)", + ) + parser.add_argument( + "--num", + type=int, + default=1, + help="Number of clients to run in parallel (default: 1)", + ) + parser.add_argument( + "--count", + type=int, + default=100, + help="Messages per client in performance mode (default: 100)", + ) + parser.add_argument( + "--cycles", + type=int, + default=2, + help="Adaptive mode cycles per client (default: 2)", + ) + parser.add_argument( + "--error", + type=int, + default=0, + help="Enable random error injection when set to 1 (default: 0)", + ) + parser.add_argument( + "--read", + type=int, + default=1, + help="Read responses when set to 1 (default: 1). Set to 0 to skip reads.", + ) + return parser.parse_args(argv) + + +def random_hex(min_len=4, max_len=64): + length = secrets.randbelow(max_len - min_len + 1) + min_len + return secrets.token_hex((length + 1) // 2)[:length] + + +def random_pad_size(min_len, max_len): + if max_len <= min_len: + return min_len + return secrets.randbelow(max_len - min_len + 1) + min_len + + +def random_pad_from_ranges(ranges): + min_len, max_len = secrets.choice(ranges) + return random_pad_size(min_len, max_len) + + +def payload_from_message(raw, data): + if raw is None: + return {"data": data} + try: + payload = json.loads(raw) + except json.JSONDecodeError: + payload = {"message": raw} + if isinstance(payload, dict): + payload["data"] = data + return payload + return {"data": data, "message": payload} + + +def _connect_client(host, port, client_id): + client = jsocket.JsonClient(address=host, port=port) + if not client.connect(): + raise RuntimeError(f"client {client_id} could not connect to {host}:{port}") + logger.info("client %s connected to %s:%s", client_id, host, port) + return client + + +def _make_payload(client_id, seq, label, pad_size): + data = random_hex() + payload = { + "client": client_id, + "seq": seq, + "label": label, + "data": data, + } + if pad_size: + payload["pad"] = "x" * pad_size + return payload, data + + +def _extract_response_data(response): + if isinstance(response, dict): + return response.get("data") + if isinstance(response, str): + return response + return None + + +def _send_and_expect(client, payload, expected_data, read_response): + client.send_obj(payload) + if not read_response: + return None + response = client.read_obj() + echoed = _extract_response_data(response) + if echoed != expected_data: + raise RuntimeError(f"unexpected response data {echoed!r} (expected {expected_data!r})") + return response + + +def _send_corrupt_payload(client, payload): + raw = json.dumps(payload, ensure_ascii=False).encode("utf-8") + mode = secrets.choice(("bad_crc", "bad_header", "invalid_json")) + if mode == "invalid_json": + raw = b"{" + checksum = zlib.crc32(raw) & 0xFFFFFFFF + if mode == "bad_crc": + checksum ^= 0xFFFFFFFF + magic = jsocket_base.FRAME_MAGIC + if mode == "bad_header": + magic = b"BAD0" + header = struct.pack(jsocket_base.FRAME_HEADER_FMT, magic, len(raw), checksum) + conn = client.conn + conn.sendall(header) + conn.sendall(raw) + + +def _send_maybe_with_error(client, payload, expected_data, enable_errors, read_response): + if enable_errors and secrets.randbelow(1000) < int(ERROR_RATE * 1000): + _send_corrupt_payload(client, payload) + return False + _send_and_expect(client, payload, expected_data, read_response) + return True + + +def _send_partial_payload(client, payload): + raw = json.dumps(payload, ensure_ascii=False).encode("utf-8") + checksum = zlib.crc32(raw) & 0xFFFFFFFF + header = struct.pack(jsocket_base.FRAME_HEADER_FMT, jsocket_base.FRAME_MAGIC, len(raw), checksum) + conn = client.conn + conn.sendall(header) + cutoff = max(1, len(raw) // 2) + conn.sendall(raw[:cutoff]) + + +def _wait_for_disconnect(client, max_wait=5.0): + deadline = time.monotonic() + max_wait + conn = client.conn + while time.monotonic() < deadline: + try: + chunk = conn.recv(1) + if not chunk: + return True + except socket.timeout: + continue + except OSError: + return True + return False + + +def _run_ping(client_id, args): + data = random_hex() + payload = payload_from_message(args.message, data) + client = _connect_client(args.host, args.port, client_id) + try: + logger.info("client %s sent %s", client_id, payload) + response = _send_and_expect(client, payload, data, args.read == 1) + finally: + client.close() + + if args.num == 1 and response is not None: + print(response) + else: + if response is not None: + logger.info("client %s received %s", client_id, response) + + +def _run_performance(client_id, args): + client = _connect_client(args.host, args.port, client_id) + start = time.monotonic() + try: + for i in range(args.count): + pad_size = random_pad_from_ranges(PERF_PAD_RANGES) + payload, data = _make_payload(client_id, i, "perf", pad_size) + ok = _send_maybe_with_error(client, payload, data, args.error == 1, args.read == 1) + if not ok: + client.close() + client = _connect_client(args.host, args.port, client_id) + if i and i % 100 == 0: + logger.info("client %s progress %s/%s", client_id, i, args.count) + finally: + client.close() + elapsed = max(time.monotonic() - start, 1e-6) + rate = args.count / elapsed + logger.info( + "client %s performance %s messages in %.2fs (%.1f msg/s)", + client_id, + args.count, + elapsed, + rate, + ) + + +def _adaptive_cycle(client_id, args, cycle_index): + seq = cycle_index * 10000 + client = _connect_client(args.host, args.port, client_id) + try: + for i in range(5): + pad_size = random_pad_size(*ADAPTIVE_SLOW_PAD_RANGE) + payload, data = _make_payload(client_id, seq, "slow", pad_size) + ok = _send_maybe_with_error(client, payload, data, args.error == 1, args.read == 1) + if not ok: + client.close() + client = _connect_client(args.host, args.port, client_id) + seq += 1 + time.sleep(1.0) + finally: + client.close() + + time.sleep(0.5) + + # Force at least one server-side recv timeout by sending a partial frame. + client = _connect_client(args.host, args.port, client_id) + try: + pad_size = random_pad_size(256, 4096) + payload, _ = _make_payload(client_id, seq, "timeout", pad_size) + _send_partial_payload(client, payload) + _wait_for_disconnect(client, max_wait=6.0) + seq += 1 + finally: + client.close() + + time.sleep(0.5) + + client = _connect_client(args.host, args.port, client_id) + try: + for i in range(25): + pad_size = random_pad_size(*ADAPTIVE_BURST_PAD_RANGE) + payload, data = _make_payload(client_id, seq, "burst", pad_size) + ok = _send_maybe_with_error(client, payload, data, args.error == 1, args.read == 1) + if not ok: + client.close() + client = _connect_client(args.host, args.port, client_id) + seq += 1 + for i in range(3): + pad_size = random_pad_size(*ADAPTIVE_LARGE_PAD_RANGE) + payload, data = _make_payload(client_id, seq, "large", pad_size) + ok = _send_maybe_with_error(client, payload, data, args.error == 1, args.read == 1) + if not ok: + client.close() + client = _connect_client(args.host, args.port, client_id) + seq += 1 + time.sleep(0.2) + finally: + client.close() + + logger.info("client %s adaptive cycle %s complete", client_id, cycle_index + 1) + + +def _run_adaptive(client_id, args): + for cycle_index in range(args.cycles): + _adaptive_cycle(client_id, args, cycle_index) + + +def _client_worker(client_id, args, results, lock): + ok = True + try: + if args.mode == "ping": + _run_ping(client_id, args) + elif args.mode == "performance": + _run_performance(client_id, args) + else: + _run_adaptive(client_id, args) + except Exception as exc: # pylint: disable=broad-exception-caught + ok = False + logger.error("client %s failed: %s", client_id, exc) + with lock: + results.append(ok) + + +def main(argv=None): + args = parse_args(argv or sys.argv[1:]) + if args.num < 1: + logger.error("--num must be >= 1") + return 2 + if args.count < 1: + logger.error("--count must be >= 1") + return 2 + if args.cycles < 1: + logger.error("--cycles must be >= 1") + return 2 + + threads = [] + results = [] + lock = threading.Lock() + for i in range(args.num): + client_id = i + 1 + t = threading.Thread( + target=_client_worker, + args=(client_id, args, results, lock), + daemon=False, + ) + threads.append(t) + t.start() + + for t in threads: + t.join() + + if not results or not all(results): + return 3 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/net_server.py b/scripts/net_server.py new file mode 100644 index 0000000..87fd7cf --- /dev/null +++ b/scripts/net_server.py @@ -0,0 +1,188 @@ +import argparse +import logging +import sys +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import jsocket +from jsocket import tserver as _tserver + + +logger = logging.getLogger("jsocket.net_server") +logging.basicConfig(level=logging.INFO, format='[%(levelname)s] %(message)s') +ANSI_RESET = "\033[0m" +ANSI_GREEN = "\033[32m" +ANSI_YELLOW = "\033[33m" +ANSI_RED = "\033[31m" + + +def _summarize_payload(obj): + data = None + pad_len = 0 + if isinstance(obj, dict): + data = obj.get("data") + pad = obj.get("pad") + if isinstance(pad, str): + pad_len = len(pad) + elif isinstance(obj, str): + data = obj + data_len = len(data) if isinstance(data, str) else 0 + return data, data_len, pad_len + + +class EchoWorker(jsocket.ServerFactoryThread): + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.timeout = 2.0 + + def _process_message(self, obj): + if isinstance(obj, dict): + client_id = obj.get("client") or obj.get("client_id") + client_id = _tserver._normalize_client_id(client_id) if client_id is not None else None + if client_id: + _tserver._set_client_identity(self, client_id) + data, data_len, pad_len = _summarize_payload(obj) + logger.info("rx data_len=%s pad_len=%s", data_len, pad_len) + return {"data": data} + + +def parse_args(argv): + parser = argparse.ArgumentParser( + description="Minimal JSON socket echo server for quick network testing.", + ) + parser.add_argument( + "host", + help="Bind address (use 0.0.0.0 to listen on all interfaces)", + ) + parser.add_argument( + "--port", + type=int, + default=5491, + help="Port to listen on (default: 5491)", + ) + return parser.parse_args(argv) + + +def format_stats(stats): + def colorize(value): + try: + num = int(value) + except (TypeError, ValueError): + return str(value) + if num == 0: + color = ANSI_GREEN + elif num <= 3: + color = ANSI_YELLOW + else: + color = ANSI_RED + return f"{color}{num}{ANSI_RESET}" + + lines = [f"client stats: connected_clients={stats.get('connected_clients', 0)}"] + clients = stats.get("clients") or {} + if not clients: + lines.append(" (no clients)") + return "\n".join(lines) + for client_id in sorted(clients.keys()): + client = clients[client_id] + display_id = client.get("client_id") or client_id + failures = client.get("failures") or {} + lines.append(f" client={display_id}") + lines.append( + " connected={connected} connects={connects} disconnects={disconnects}".format( + connected=client.get("connected"), + connects=client.get("connects"), + disconnects=client.get("disconnects"), + ) + ) + lines.append( + " messages_in={messages_in} messages_out={messages_out}".format( + messages_in=client.get("messages_in"), + messages_out=client.get("messages_out"), + ) + ) + lines.append( + " bytes_in={bytes_in} bytes_out={bytes_out}".format( + bytes_in=client.get("bytes_in"), + bytes_out=client.get("bytes_out"), + ) + ) + lines.append( + " avg_payload_in={avg_in:.2f} avg_payload_out={avg_out:.2f}".format( + avg_in=client.get("avg_payload_in", 0.0), + avg_out=client.get("avg_payload_out", 0.0), + ) + ) + lines.append( + " failures timeout={timeout} bad_write={bad_write} bad_crc={bad_crc} bad_header={bad_header}".format( + timeout=colorize(failures.get("timeout", 0)), + bad_write=colorize(failures.get("bad_write", 0)), + bad_crc=colorize(failures.get("bad_crc", 0)), + bad_header=colorize(failures.get("bad_header", 0)), + ) + ) + lines.append( + " failures oversize={oversize} invalid_utf8={invalid_utf8} invalid_json={invalid_json}".format( + oversize=colorize(failures.get("oversize", 0)), + invalid_utf8=colorize(failures.get("invalid_utf8", 0)), + invalid_json=colorize(failures.get("invalid_json", 0)), + ) + ) + lines.append( + " failures handler={handler} framing={framing}".format( + handler=colorize(failures.get("handler", 0)), + framing=colorize(failures.get("framing", 0)), + ) + ) + lines.append( + " last_connect_ts={last_connect_ts} last_disconnect_ts={last_disconnect_ts}".format( + last_connect_ts=client.get("last_connect_ts"), + last_disconnect_ts=client.get("last_disconnect_ts"), + ) + ) + lines.append( + " last_message_ts={last_message_ts} connected_duration={connected_duration:.2f}".format( + last_message_ts=client.get("last_message_ts"), + connected_duration=client.get("connected_duration", 0.0), + ) + ) + lines.append( + " total_connected_duration={total_connected_duration:.2f}".format( + total_connected_duration=client.get("total_connected_duration", 0.0), + ) + ) + return "\n".join(lines) + + +def main(argv=None): + args = parse_args(argv or sys.argv[1:]) + + try: + server = jsocket.ServerFactory(EchoWorker, address=args.host, port=args.port) + except OSError as exc: + logger.error("failed to bind %s:%s (%s)", args.host, args.port, exc) + return 1 + + server.start() + logger.info("listening on %s:%s", args.host, args.port) + try: + next_report = time.monotonic() + 10.0 + while True: + time.sleep(0.5) + now = time.monotonic() + if now >= next_report: + logger.info("%s", format_stats(server.get_client_stats())) + next_report = now + 10.0 + except KeyboardInterrupt: + logger.info("shutting down") + finally: + server.stop() + server.join(timeout=3) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_additional_coverage.py b/tests/test_additional_coverage.py index e21ed8e..63c4382 100644 --- a/tests/test_additional_coverage.py +++ b/tests/test_additional_coverage.py @@ -2,6 +2,10 @@ These tests avoid real network I/O by monkeypatching/stubbing where possible. """ +# pylint: disable=protected-access,missing-function-docstring,missing-class-docstring +# pylint: disable=too-few-public-methods,non-parent-init-called,super-init-not-called +# pylint: disable=too-many-arguments,too-many-positional-arguments,unused-argument +# pylint: disable=attribute-defined-outside-init,useless-return import threading import time @@ -9,7 +13,7 @@ import pytest import jsocket -import jsocket.tserver as tserver +from jsocket import jsocket_base, tserver def test_client_connect_failure_returns_false(monkeypatch): @@ -19,7 +23,7 @@ def fail_connect(self, addr): # pylint: disable=unused-argument raise OSError("boom") # Short-circuit sleeps so the test is fast - monkeypatch.setattr(jsocket.jsocket_base.time, "sleep", lambda *_: None) + monkeypatch.setattr(jsocket_base.time, "sleep", lambda *_: None) monkeypatch.setattr(socket.socket, "connect", fail_connect, raising=True) try: @@ -74,10 +78,10 @@ def fake_socket(*args, **kwargs): # pylint: disable=unused-argument def no_sleep(*args, **kwargs): # pylint: disable=unused-argument raise AssertionError("unexpected connect backoff sleep") - monkeypatch.setattr(jsocket.jsocket_base.socket, "socket", fake_socket, raising=True) - monkeypatch.setattr(jsocket.jsocket_base.time, "sleep", no_sleep, raising=True) + monkeypatch.setattr(jsocket_base.socket, "socket", fake_socket, raising=True) + monkeypatch.setattr(jsocket_base.time, "sleep", no_sleep, raising=True) - client = jsocket.jsocket_base.JsonClient.__new__(jsocket.jsocket_base.JsonClient) + client = jsocket_base.JsonClient.__new__(jsocket_base.JsonClient) client.socket = ClosedSocket() client.conn = client.socket client._recv_timeout = 0.25 diff --git a/tests/test_integration_failures.py b/tests/test_integration_failures.py index cd06a80..b8ff6c2 100644 --- a/tests/test_integration_failures.py +++ b/tests/test_integration_failures.py @@ -8,7 +8,7 @@ import pytest import jsocket -import jsocket.jsocket_base as jsocket_base +from jsocket import jsocket_base class EchoServer(jsocket.ThreadedServer): diff --git a/tests/test_server_stats.py b/tests/test_server_stats.py index f127e28..12a7769 100644 --- a/tests/test_server_stats.py +++ b/tests/test_server_stats.py @@ -56,13 +56,22 @@ def test_threadedserver_client_stats(): stats = server.get_client_stats() assert stats["connected_clients"] == 1 assert len(stats["clients"]) == 1 - duration = next(iter(stats["clients"].values())) - assert duration >= 0.0 + client_stats = next(iter(stats["clients"].values())) + assert client_stats["connected"] is True + assert client_stats["messages_in"] >= 1 + assert client_stats["messages_out"] >= 1 + assert client_stats["last_connect_ts"] is not None + assert client_stats["last_message_ts"] is not None + assert client_stats["avg_payload_in"] > 0.0 + assert client_stats["avg_payload_out"] > 0.0 client.close() time.sleep(0.2) stats_after = server.get_client_stats() assert stats_after["connected_clients"] == 0 + client_stats_after = next(iter(stats_after["clients"].values())) + assert client_stats_after["connected"] is False + assert client_stats_after["disconnects"] >= 1 finally: if client is not None: try: @@ -104,7 +113,11 @@ def test_serverfactory_client_stats(): stats = server.get_client_stats() assert stats["connected_clients"] == 2 assert len(stats["clients"]) == 2 - assert all(duration >= 0.0 for duration in stats["clients"].values()) + for client_stats in stats["clients"].values(): + assert client_stats["connected"] is True + assert client_stats["messages_in"] >= 1 + assert client_stats["messages_out"] >= 1 + assert client_stats["last_message_ts"] is not None finally: if c1 is not None: try: diff --git a/tests/test_unit_coverage.py b/tests/test_unit_coverage.py index 505e5f2..d1d84e6 100644 --- a/tests/test_unit_coverage.py +++ b/tests/test_unit_coverage.py @@ -1,14 +1,16 @@ """Unit tests to cover error/edge branches without real network I/O.""" +# pylint: disable=protected-access,missing-function-docstring,missing-class-docstring +# pylint: disable=too-few-public-methods,non-parent-init-called,super-init-not-called +# pylint: disable=too-many-arguments,too-many-positional-arguments,unused-argument +# pylint: disable=useless-parent-delegation,too-many-instance-attributes +# pylint: disable=attribute-defined-outside-init,use-implicit-booleaness-not-comparison import threading -import socket import struct import zlib import pytest -import jsocket -import jsocket.jsocket_base as jsocket_base -import jsocket.tserver as tserver +from jsocket import jsocket_base, tserver class FakeSocket: # pylint: disable=missing-function-docstring @@ -722,3 +724,4 @@ def test_serverfactory_get_client_stats_active_threads(): stats = factory.get_client_stats() assert stats["connected_clients"] == 2 assert len(stats["clients"]) == 2 + assert all(client_stats["connected"] is True for client_stats in stats["clients"].values())