From fa40ddb10ba881151387a92a3875c1aa7bd33e0e Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Thu, 23 Jul 2026 11:20:08 +0800 Subject: [PATCH 1/8] Bound external process calls with explicit timeouts ps / tasklist / taskkill and the git shard-split invocation ran without a timeout, so a wedged subprocess blocked the calling thread indefinitely. Broaden the handlers from CalledProcessError to SubprocessError so the newly reachable TimeoutExpired is caught rather than escaping. --- .../utils/process_supervisor/supervisor.py | 14 +++++++++++--- je_web_runner/utils/sharding/diff_shard.py | 7 ++++++- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/je_web_runner/utils/process_supervisor/supervisor.py b/je_web_runner/utils/process_supervisor/supervisor.py index 3b154669..9e0a40b5 100644 --- a/je_web_runner/utils/process_supervisor/supervisor.py +++ b/je_web_runner/utils/process_supervisor/supervisor.py @@ -51,6 +51,11 @@ class OrphanFinding: ProcessLister = Callable[[], list[OrphanFinding]] ProcessKiller = Callable[[int], bool] +# Every external process call is bounded: a wedged ``ps`` / ``tasklist`` / +# ``taskkill`` would otherwise block the calling thread forever, which +# defeats the point of a supervisor meant to unstick hung runs. +_PROCESS_CALL_TIMEOUT_SECONDS = 15.0 + def _ps_unix_lister() -> list[OrphanFinding]: try: @@ -59,8 +64,9 @@ def _ps_unix_lister() -> list[OrphanFinding]: ["ps", "-Ao", "pid=,comm=,args="], text=True, stderr=subprocess.DEVNULL, + timeout=_PROCESS_CALL_TIMEOUT_SECONDS, ) - except (FileNotFoundError, subprocess.CalledProcessError) as error: + except (OSError, subprocess.SubprocessError) as error: raise ProcessSupervisorError(f"ps failed: {error!r}") from error findings: list[OrphanFinding] = [] for line in out.splitlines(): @@ -84,8 +90,9 @@ def _tasklist_windows_lister() -> list[OrphanFinding]: ["tasklist", "/FO", "CSV", "/NH"], text=True, stderr=subprocess.DEVNULL, + timeout=_PROCESS_CALL_TIMEOUT_SECONDS, ) - except (FileNotFoundError, subprocess.CalledProcessError) as error: + except (OSError, subprocess.SubprocessError) as error: raise ProcessSupervisorError(f"tasklist failed: {error!r}") from error findings: list[OrphanFinding] = [] for line in out.splitlines(): @@ -116,6 +123,7 @@ def default_killer(pid: int) -> bool: ["taskkill", "/F", "/PID", str(pid)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + timeout=_PROCESS_CALL_TIMEOUT_SECONDS, ) return True # The PID list is filtered by ``KNOWN_DRIVER_NAMES`` and excludes @@ -123,7 +131,7 @@ def default_killer(pid: int) -> bool: # supervisor's own webdriver children. os.kill(pid, 9) # NOSONAR S4828 — pid pre-validated against driver name allow-list return True - except (OSError, subprocess.CalledProcessError) as error: + except (OSError, subprocess.SubprocessError) as error: web_runner_logger.warning(f"process_supervisor kill {pid} failed: {error!r}") return False diff --git a/je_web_runner/utils/sharding/diff_shard.py b/je_web_runner/utils/sharding/diff_shard.py index 2dea1e05..0fa47c64 100644 --- a/je_web_runner/utils/sharding/diff_shard.py +++ b/je_web_runner/utils/sharding/diff_shard.py @@ -21,6 +21,8 @@ class DiffShardError(WebRunnerException): GitRunner = Callable[[Sequence[str]], str] +_GIT_TIMEOUT_SECONDS = 30.0 + def _default_git_runner(args: Sequence[str]) -> str: cmd = ["git", *args] @@ -30,8 +32,11 @@ def _default_git_runner(args: Sequence[str]) -> str: cmd, stderr=subprocess.DEVNULL, text=True, + # Bounded so a git subprocess waiting on an index lock or a + # credential prompt can't wedge the shard split indefinitely. + timeout=_GIT_TIMEOUT_SECONDS, ) - except (FileNotFoundError, subprocess.CalledProcessError) as error: + except (OSError, subprocess.SubprocessError) as error: raise DiffShardError(f"git command failed: {error!r}") from error return out From 9ff9669e5439d9c78475fea61f2de38f7b4a6fe5 Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Thu, 23 Jul 2026 11:20:20 +0800 Subject: [PATCH 2/8] Bound BrowserPool checkout retries on unhealthy sessions Destroying an unhealthy session frees a tracked slot, so _can_grow() flipped straight back to True and checkout() spawned again. A factory that always produced unhealthy instances (crash-on-launch browser, dead health check) looped forever, churning real browser processes and ignoring the timeout. Cap the replacement attempts at size * 3 so the failure is deterministic rather than a spin that only stops at the deadline. --- je_web_runner/utils/browser_pool/pool.py | 29 ++++++++++++++++++++---- test/unit_test/test_browser_pool.py | 17 ++++++++++++++ 2 files changed, 41 insertions(+), 5 deletions(-) diff --git a/je_web_runner/utils/browser_pool/pool.py b/je_web_runner/utils/browser_pool/pool.py index 1ca4da4c..be8d355a 100644 --- a/je_web_runner/utils/browser_pool/pool.py +++ b/je_web_runner/utils/browser_pool/pool.py @@ -89,12 +89,31 @@ def checkout(self, timeout: float = 30.0) -> PooledSession: if self._closed: raise BrowserPoolError("pool is closed") deadline = time.monotonic() + timeout - while True: + # Discarding an unhealthy session frees a ``_tracked`` slot, so + # ``_can_grow()`` flips straight back to True and the next iteration + # spawns again. A factory that always produces unhealthy instances + # (crash-on-launch browser, dead health-check endpoint) would + # otherwise loop forever, churning real browser processes. Bound the + # retries so the failure is fast and deterministic rather than a + # spin that only stops at the deadline. + budget = self._unhealthy_retry_budget() + for _ in range(budget): session = self._acquire_session(timeout, deadline) - if not self._is_healthy(session): - self._destroy(session) - continue - return session + if self._is_healthy(session): + return session + self._destroy(session) + if time.monotonic() >= deadline: + raise BrowserPoolError( + f"no healthy session available within {timeout}s" + ) + raise BrowserPoolError( + f"no healthy session after {budget} attempts; " + f"the factory or health_check is failing consistently" + ) + + def _unhealthy_retry_budget(self) -> int: + """Allow every pooled slot a couple of replacement attempts.""" + return max(4, self._size * 3) def _acquire_session(self, timeout: float, deadline: float) -> PooledSession: try: diff --git a/test/unit_test/test_browser_pool.py b/test/unit_test/test_browser_pool.py index bbc05069..94f64a43 100644 --- a/test/unit_test/test_browser_pool.py +++ b/test/unit_test/test_browser_pool.py @@ -80,6 +80,23 @@ def health(_instance): self.assertNotEqual(sess1.session_id, sess2.session_id) destructor.assert_called() # destroyed at least once + def test_permanently_unhealthy_factory_times_out(self): + """A factory whose instances never pass the health check must make + checkout() give up at the deadline instead of spinning forever + spawning-and-destroying browser processes.""" + spawned = MagicMock(side_effect=lambda: object()) + pool = BrowserPool( + factory=spawned, + destructor=MagicMock(), + health_check=lambda _instance: False, + size=2, + ) + with self.assertRaises(BrowserPoolError): + pool.checkout(timeout=30.0) + # Deterministic bound: size(2) * 3 attempts, no clock dependency and + # no unbounded churn of real browser processes. + self.assertEqual(spawned.call_count, 6) + def test_factory_failure_raises(self): def failing(): raise RuntimeError("no driver") From 375d5378017cfa481cf67b8de7bc6ac442705137 Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Thu, 23 Jul 2026 11:20:20 +0800 Subject: [PATCH 3/8] Harden WebdriverManager cleanup against partial failures quit() aborted its loop on the first driver that raised, leaving the remaining browsers running as orphans and the tracking list uncleared. Attempt every driver and report the failures afterwards. The close paths left current_webdriver pointing at a closed session, and neither path cleared the shared webdriver_wrapper singleton, so the next caller inherited a dead driver and a stale ActionChains. Close before dropping the reference too, so a failed close stays tracked for quit(). --- je_web_runner/manager/webrunner_manager.py | 56 ++++++++-- test/unit_test/test_webrunner_manager.py | 124 +++++++++++++++++++++ 2 files changed, 172 insertions(+), 8 deletions(-) diff --git a/je_web_runner/manager/webrunner_manager.py b/je_web_runner/manager/webrunner_manager.py index 1155e48b..f3c1ea67 100644 --- a/je_web_runner/manager/webrunner_manager.py +++ b/je_web_runner/manager/webrunner_manager.py @@ -76,6 +76,17 @@ def change_webdriver(self, index_of_webdriver: int) -> None: f"WebdriverManager change_webdriver, index_of_webdriver: {index_of_webdriver}, failed: {error!r}") record_action_to_list("web runner manager change_webdriver", param, error) + def _detach_wrapper_if_active(self, driver: WebDriver) -> None: + """ + 若共用的 wrapper 單例正指向這個已關閉的 driver，就把它解除綁定。 + Clear the shared ``webdriver_wrapper`` singleton when it is still + pointing at ``driver``. The manager owns raw Selenium handles, so + closing one here would otherwise leave the wrapper (and its + ActionChains) bound to a dead session for the next caller. + """ + if self.webdriver_wrapper.current_webdriver is driver: + self.webdriver_wrapper.set_active_driver(None) + def close_current_webdriver(self) -> None: """ 關閉當前 WebDriver @@ -83,8 +94,18 @@ def close_current_webdriver(self) -> None: """ web_runner_logger.info("WebdriverManager close_current_webdriver") try: - self._current_webdriver_list.remove(self.current_webdriver) - self.current_webdriver.close() + if self.current_webdriver is None: + raise WebRunnerWebDriverIsNoneException(selenium_wrapper_web_driver_not_found_error) + # Close first, then drop the reference: if close() raises, the + # driver stays tracked so a later quit() can still reclaim it. + closed = self.current_webdriver + closed.close() + if closed in self._current_webdriver_list: + self._current_webdriver_list.remove(closed) + # Never leave ``current_webdriver`` pointing at a closed session — + # the next action would fail against a dead driver. + self.current_webdriver = None + self._detach_wrapper_if_active(closed) record_action_to_list("web runner manager close_current_webdriver", None, None) except Exception as error: web_runner_logger.error(f"WebdriverManager close_current_webdriver, failed: {error!r}") @@ -100,9 +121,12 @@ def close_choose_webdriver(self, webdriver_index: int) -> None: web_runner_logger.info("WebdriverManager close_choose_webdriver") param = locals() try: - self.current_webdriver = self._current_webdriver_list[webdriver_index] - self.current_webdriver.close() - self._current_webdriver_list.remove(self._current_webdriver_list[webdriver_index]) + chosen = self._current_webdriver_list[webdriver_index] + chosen.close() + self._current_webdriver_list.pop(webdriver_index) + if self.current_webdriver is chosen: + self.current_webdriver = None + self._detach_wrapper_if_active(chosen) record_action_to_list("web runner manager close_choose_webdriver", param, None) except Exception as error: web_runner_logger.error(f"WebdriverManager close_choose_webdriver, failed: {error!r}") @@ -122,12 +146,28 @@ def quit(self) -> None: # Clean test records test_object_record.clean_record() - # 關閉所有 WebDriver - # Quit all WebDrivers + # 關閉所有 WebDriver；單一 driver 失敗不可中斷其餘的清理， + # 否則剩下的瀏覽器程序會變成 orphan。 + # Quit every WebDriver. A failure on one must not abort the loop, + # or the remaining browsers leak as orphan processes. + quit_errors: list[Exception] = [] for webdriver in self._current_webdriver_list: - webdriver.quit() + try: + webdriver.quit() + except Exception as error: # pylint: disable=broad-except + web_runner_logger.error(f"WebdriverManager quit driver failed: {error!r}") + quit_errors.append(error) self._current_webdriver_list = [] + self.current_webdriver = None + # Every driver is gone, so the shared wrapper singleton must not + # keep serving a dead one to the next caller. + self.webdriver_wrapper.set_active_driver(None) + if quit_errors: + raise WebDriverException( + f"WebdriverManager quit failed for {len(quit_errors)} driver(s): " + f"{[repr(e) for e in quit_errors]}" + ) record_action_to_list("web runner manager quit", None, None) except Exception as error: web_runner_logger.error(f"WebdriverManager quit, failed: {error!r}") diff --git a/test/unit_test/test_webrunner_manager.py b/test/unit_test/test_webrunner_manager.py index 61ae85ff..48162661 100644 --- a/test/unit_test/test_webrunner_manager.py +++ b/test/unit_test/test_webrunner_manager.py @@ -2,6 +2,8 @@ import unittest from unittest.mock import MagicMock +from selenium.common.exceptions import WebDriverException + from je_web_runner.manager.webrunner_manager import WebdriverManager from je_web_runner.webdriver.webdriver_wrapper import webdriver_wrapper_instance @@ -55,5 +57,127 @@ def test_set_active_driver_none_clears_chain(self): self.assertIsNone(webdriver_wrapper_instance._action_chain) +class TestWrapperSingletonIsDetached(unittest.TestCase): + """The manager owns raw Selenium handles, so closing one must also clear + the shared wrapper singleton — otherwise it keeps serving a dead driver + (and a stale ActionChains) to whoever calls next.""" + + def setUp(self): + self._saved_driver = webdriver_wrapper_instance.current_webdriver + self._saved_chain = webdriver_wrapper_instance._action_chain + + def tearDown(self): + webdriver_wrapper_instance.current_webdriver = self._saved_driver + webdriver_wrapper_instance._action_chain = self._saved_chain + + def test_quit_detaches_wrapper(self): + manager = WebdriverManager() + driver = MagicMock(name="driver") + manager._current_webdriver_list = [driver] + webdriver_wrapper_instance.set_active_driver(driver) + + manager.quit() + + self.assertIsNone(webdriver_wrapper_instance.current_webdriver) + self.assertIsNone(webdriver_wrapper_instance._action_chain) + + def test_close_current_detaches_wrapper(self): + manager = WebdriverManager() + driver = MagicMock(name="driver") + manager._current_webdriver_list = [driver] + manager.current_webdriver = driver + webdriver_wrapper_instance.set_active_driver(driver) + + manager.close_current_webdriver() + + self.assertIsNone(webdriver_wrapper_instance.current_webdriver) + + def test_close_choose_leaves_unrelated_active_driver_alone(self): + manager = WebdriverManager() + driver0, driver1 = MagicMock(name="driver0"), MagicMock(name="driver1") + manager._current_webdriver_list = [driver0, driver1] + # The wrapper is driving driver0; closing driver1 must not disturb it. + webdriver_wrapper_instance.set_active_driver(driver0) + + manager.close_choose_webdriver(1) + + self.assertIs(webdriver_wrapper_instance.current_webdriver, driver0) + + +class TestQuitIsResilient(unittest.TestCase): + """One driver failing to quit must not strand the remaining browsers.""" + + def test_quit_continues_past_failing_driver(self): + manager = WebdriverManager() + bad = MagicMock(name="bad") + bad.quit.side_effect = RuntimeError("driver already dead") + good_before, good_after = MagicMock(name="before"), MagicMock(name="after") + manager._current_webdriver_list = [good_before, bad, good_after] + + with self.assertRaises(WebDriverException): + manager.quit() + + # Every driver gets a quit() attempt, including the one queued after + # the failure — otherwise it leaks as an orphan browser process. + good_before.quit.assert_called_once() + good_after.quit.assert_called_once() + self.assertEqual(manager._current_webdriver_list, []) + self.assertIsNone(manager.current_webdriver) + + def test_quit_clears_state_on_success(self): + manager = WebdriverManager() + driver = MagicMock(name="driver") + manager._current_webdriver_list = [driver] + manager.current_webdriver = driver + + manager.quit() + + driver.quit.assert_called_once() + self.assertEqual(manager._current_webdriver_list, []) + self.assertIsNone(manager.current_webdriver) + + +class TestCloseClearsCurrentDriver(unittest.TestCase): + """A closed driver must never stay installed as ``current_webdriver``.""" + + def test_close_current_webdriver_clears_reference(self): + manager = WebdriverManager() + driver = MagicMock(name="driver") + manager._current_webdriver_list = [driver] + manager.current_webdriver = driver + + manager.close_current_webdriver() + + driver.close.assert_called_once() + self.assertIsNone(manager.current_webdriver) + self.assertEqual(manager._current_webdriver_list, []) + + def test_failed_close_keeps_driver_tracked(self): + manager = WebdriverManager() + driver = MagicMock(name="driver") + driver.close.side_effect = RuntimeError("close failed") + manager._current_webdriver_list = [driver] + manager.current_webdriver = driver + + manager.close_current_webdriver() + + # close() blew up, so the driver stays tracked and a later quit() + # can still reclaim the process. + self.assertEqual(manager._current_webdriver_list, [driver]) + + def test_close_choose_webdriver_removes_only_that_index(self): + manager = WebdriverManager() + driver0, driver1 = MagicMock(name="driver0"), MagicMock(name="driver1") + manager._current_webdriver_list = [driver0, driver1] + manager.current_webdriver = driver1 + + manager.close_choose_webdriver(1) + + driver1.close.assert_called_once() + driver0.close.assert_not_called() + self.assertEqual(manager._current_webdriver_list, [driver0]) + self.assertIsNone(manager.current_webdriver) + + if __name__ == "__main__": unittest.main() From 64412eed78f4d99577cfbab18fb6cda212dc402e Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Thu, 23 Jul 2026 11:20:47 +0800 Subject: [PATCH 4/8] Add length-prefixed framing to the socket server The handler did a single recv(8192), so any request larger than one segment was silently truncated, and a client that connected without sending pinned its thread forever (one thread per connection under ThreadingMixIn). Add a receive timeout and read until the request is actually complete. New clients can opt into WRLEN \n framing, which is auto-detected, so existing clients keep working unchanged and replies mirror the dialect the request arrived in. Framing also fixes authentication when the token line straddles a packet boundary, which the legacy heuristic cannot resolve. Legacy requests fall back to an incremental bracket-balance scanner that distinguishes a truncated body from a malformed one, keeping the prompt error reply for bad JSON. Requests are capped at 8 MiB. Export send_command / encode_frame / read_frame so the framed format is usable from client code. --- je_web_runner/__init__.py | 4 + .../socket_server/web_runner_socket_server.py | 353 +++++++++++++++++- test/unit_test/test_socket_server.py | 255 ++++++++++++- 3 files changed, 595 insertions(+), 17 deletions(-) diff --git a/je_web_runner/__init__.py b/je_web_runner/__init__.py index c916e463..be6e7cdc 100644 --- a/je_web_runner/__init__.py +++ b/je_web_runner/__init__.py @@ -24,6 +24,9 @@ from je_web_runner.utils.generate_report.generate_junit_xml_report import generate_junit_xml_report from je_web_runner.utils.generate_report.generate_allure_report import generate_allure from je_web_runner.utils.generate_report.generate_allure_report import generate_allure_report +from je_web_runner.utils.socket_server.web_runner_socket_server import encode_frame +from je_web_runner.utils.socket_server.web_runner_socket_server import read_frame +from je_web_runner.utils.socket_server.web_runner_socket_server import send_command from je_web_runner.utils.socket_server.web_runner_socket_server import start_web_runner_socket_server from je_web_runner.utils.test_object.test_object_class import TestObject from je_web_runner.utils.test_object.test_object_class import create_test_object @@ -387,6 +390,7 @@ "generate_junit_xml", "generate_junit_xml_report", "generate_allure", "generate_allure_report", "start_web_runner_socket_server", "get_dir_files_as_list", + "send_command", "read_frame", "encode_frame", "TestObject", "create_test_object", "get_test_object_type_list", "test_record_instance", "Keys", "callback_executor", "create_project_dir", "load_env", "get_env", "expand_in_action", "EnvConfigError", diff --git a/je_web_runner/utils/socket_server/web_runner_socket_server.py b/je_web_runner/utils/socket_server/web_runner_socket_server.py index 46ccb071..5963c76d 100644 --- a/je_web_runner/utils/socket_server/web_runner_socket_server.py +++ b/je_web_runner/utils/socket_server/web_runner_socket_server.py @@ -1,3 +1,4 @@ +import codecs import json import socketserver import ssl @@ -11,6 +12,73 @@ _NEWLINE = b"\n" _QUIT_COMMAND = "quit_server" _RECV_BYTES = 8192 +# A connected client that never sends anything would otherwise pin its +# handler thread forever; ThreadingMixIn spawns one thread per connection, +# so a handful of idle sockets could exhaust the server. +_RECV_TIMEOUT_SECONDS = 30.0 +# Upper bound on a single request so a peer streaming endlessly cannot grow +# the handler's buffer without limit. +_MAX_REQUEST_BYTES = 8 * 1024 * 1024 + +# --------------------------------------------------------------------------- +# 長度前綴框架 / Length-prefixed framing +# +# 舊協定沒有長度資訊,伺服器只能用括號平衡「猜」訊息是否收完,而且 token 那一行 +# 若跨越封包邊界就無法可靠切出。加上長度前綴後這兩個問題都消失。 +# +# The legacy protocol carries no length, so the server has to *infer* where a +# message ends from bracket balance, and a token line split across packet +# boundaries cannot be separated reliably. A length prefix removes both +# problems: +# +# WRLEN \n +# +# ``body`` is byte-for-byte what the legacy protocol sent (optional +# ``token\n`` line, then the JSON command), so framing is purely a transport +# concern layered *under* auth and parsing. +# +# Framing is auto-detected, never required: a request that does not open with +# the marker is served exactly as before, so existing clients keep working. +# Replies mirror the request's dialect — a framed request gets a framed reply, +# a legacy request keeps the ``Return_Data_Over_JE`` sentinel. +# --------------------------------------------------------------------------- +_LENGTH_PREFIX = b"WRLEN " +# marker + decimal digits + newline; anything longer is not a valid header. +_MAX_HEADER_BYTES = len(_LENGTH_PREFIX) + 24 + +_MODE_FRAMED = "framed" +_MODE_LEGACY = "legacy" +_MODE_UNDECIDED = "undecided" + + +def _detect_mode(buffer: bytes) -> str: + """ + 從已收到的位元組判斷這是不是長度前綴請求。 + Decide whether ``buffer`` opens a framed request. + + Returns ``_MODE_UNDECIDED`` while the bytes so far are still a proper + prefix of the marker (a tiny first ``recv`` must not be misread as + legacy). Any other divergence settles it immediately, which keeps the + legacy rejection path as prompt as it has always been. + """ + if buffer.startswith(_LENGTH_PREFIX): + return _MODE_FRAMED + if _LENGTH_PREFIX.startswith(buffer): + return _MODE_UNDECIDED + return _MODE_LEGACY + + +def encode_frame(body: bytes) -> bytes: + """ + 將 ``body`` 包成長度前綴訊框。 + Wrap ``body`` in a length-prefixed frame. + + :param body: 要傳送的原始位元組 / raw bytes to send + :return: 可直接寫入 socket 的訊框 / a frame ready to write to the socket + """ + if not isinstance(body, (bytes, bytearray)): + raise TypeError("body must be bytes") + return _LENGTH_PREFIX + str(len(body)).encode("ascii") + _NEWLINE + bytes(body) def _split_token(payload: bytes) -> tuple[str | None, bytes]: @@ -21,6 +89,68 @@ def _split_token(payload: bytes) -> tuple[str | None, bytes]: return parts[0].decode("utf-8", errors="replace"), parts[1] +class _RequestFrameScanner: + """ + 以遞增方式判斷請求是否已收完(用來決定還要不要繼續讀 socket)。 + Incremental stop condition for reading one request off the stream. + + The wire protocol carries no length prefix, so the handler cannot know + how many ``recv`` calls a single request spans. Bracket/brace balance is + a reliable signal: an unbalanced document is still in flight, a balanced + one is finished (even if it later fails to parse — in that case the + caller should reply with an error rather than block waiting for bytes + that will never arrive). + + Each byte is scanned exactly once. Feeding the whole buffer on every + ``recv`` instead would be quadratic in the request size. Decoding is + incremental too, so a multi-byte character split across a chunk boundary + is not mangled into a spurious delimiter. + """ + + def __init__(self) -> None: + self._decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + self._depth = 0 + self._in_string = False + self._escaped = False + self._saw_content = False + self._unbalanced_close = False + + def feed(self, chunk: bytes) -> None: + for char in self._decoder.decode(chunk): + self._consume(char) + + def _consume(self, char: str) -> None: + if self._in_string: + if self._escaped: + self._escaped = False + elif char == "\\": + self._escaped = True + elif char == '"': + self._in_string = False + return + if char == '"': + self._in_string = True + self._saw_content = True + elif char in "[{": + self._depth += 1 + self._saw_content = True + elif char in "]}": + self._depth -= 1 + if self._depth < 0: + # More closers than openers: malformed, not truncated. + self._unbalanced_close = True + elif not char.isspace(): + self._saw_content = True + + @property + def is_complete(self) -> bool: + if self._unbalanced_close: + return True + if self._in_string: + return False + return self._saw_content and self._depth == 0 + + def _send_chunks(send, address, *chunks: bytes) -> None: """Send several byte chunks to a UDP-style endpoint, each with a trailing newline.""" for chunk in chunks: @@ -34,6 +164,26 @@ class TCPServerHandler(socketserver.BaseRequestHandler): Request handler for TCP server """ + # Set per request in ``handle``; declared here so the attribute always + # exists even if a subclass reaches a reply path early. + _framed = False + + def _reply(self, *chunks: bytes) -> None: + """ + 以請求所用的格式回覆(長度前綴或舊的結束標記)。 + Reply in whichever dialect the request arrived in: one length-prefixed + frame for framed clients, newline-separated chunks terminated by + ``Return_Data_Over_JE`` for legacy ones. + """ + try: + if self._framed: + self.request.sendall(encode_frame(_NEWLINE.join(chunks))) + return + _send_chunks(self.request.sendto, self.client_address, *chunks) + self.request.sendto(_END_MARKER, self.client_address) + except OSError as send_error: + print(repr(send_error)) + def _authorize(self, payload: bytes) -> bytes | None: """Return the authenticated payload, or None to reject the request.""" expected = getattr(self.server, "auth_token", None) @@ -41,12 +191,16 @@ def _authorize(self, payload: bytes) -> bytes | None: return payload token, remainder = _split_token(payload) if token is None or not compare_digest(token, expected): - self.request.sendto(b"unauthorized\n", self.client_address) - self.request.sendto(_END_MARKER, self.client_address) + self._reply(b"unauthorized") return None return remainder.strip() def _handle_quit(self) -> None: + # A framed client blocks reading its reply frame, so acknowledge + # before tearing the server down. Legacy clients never got a reply + # here, so that path stays byte-identical. + if self._framed: + self._reply(b"quit_server") self.server.shutdown() self.server.close_flag = True # Wake any waiter blocked on ``close_event.wait(timeout=...)``. @@ -56,26 +210,130 @@ def _handle_quit(self) -> None: def _execute_and_reply(self, command_string: str) -> None: try: execute_str = json.loads(command_string) - socket = self.request - for execute_return in execute_action(execute_str).values(): - _send_chunks(socket.sendto, self.client_address, str(execute_return).encode("utf-8")) - socket.sendto(_END_MARKER, self.client_address) + returns = [ + str(execute_return).encode("utf-8") + for execute_return in execute_action(execute_str).values() + ] except (ValueError, TypeError, OSError) as error: # JSONDecodeError inherits from ValueError, so it is already # covered above (SonarCloud S5713). - try: - socket = self.request - _send_chunks(socket.sendto, self.client_address, str(error).encode("utf-8")) - socket.sendto(_END_MARKER, self.client_address) - except OSError as send_error: - print(repr(send_error)) + self._reply(str(error).encode("utf-8")) + return + self._reply(*returns) - def handle(self): - raw = self.request.recv(_RECV_BYTES).strip() - payload = self._authorize(raw) + def _recv_chunk(self) -> bytes | None: + """One bounded ``recv``; ``None`` signals the socket is unusable.""" + try: + return self.request.recv(_RECV_BYTES) + except OSError as error: + # Covers socket.timeout (an OSError subclass) and peers that + # reset the connection before sending a full request. + print(f"socket receive failed: {error!r}", flush=True) + return None + + def _fill_until(self, buffer: bytearray, predicate, limit: int) -> bool: + """Keep reading into ``buffer`` until ``predicate`` holds or ``limit`` trips.""" + while not predicate(buffer): + if len(buffer) > limit: + print(f"request exceeded {limit} bytes; dropping", flush=True) + return False + chunk = self._recv_chunk() + if not chunk: + return False + buffer.extend(chunk) + return True + + def _read_framed_body(self, buffer: bytearray) -> bytes | None: + """ + 讀取 ``WRLEN \\n`` 標頭並收滿 n 個位元組。 + Read the ``WRLEN \\n`` header, then exactly ``n`` bytes of body. + + With an explicit length there is no guessing: the body is complete + when the byte count is met, so a token line spanning packet + boundaries is no longer a problem. + """ + if not self._fill_until(buffer, lambda buf: _NEWLINE in buf, _MAX_HEADER_BYTES): + return None + header, _, rest = bytes(buffer).partition(_NEWLINE) + try: + declared = int(header[len(_LENGTH_PREFIX):].strip()) + except ValueError: + print(f"malformed length header: {header!r}", flush=True) + return None + if declared < 0 or declared > _MAX_REQUEST_BYTES: + print(f"declared length out of range: {declared}", flush=True) + return None + body = bytearray(rest) + if not self._fill_until(body, lambda buf: len(buf) >= declared, _MAX_REQUEST_BYTES): + # Peer stopped early: an under-length body is a truncated + # request, not something to hand to the executor. + print( + f"framed body truncated: got {len(body)} of {declared} bytes", + flush=True, + ) + return None + return bytes(body[:declared]) + + def _read_legacy_body(self, buffer: bytearray) -> bytes | None: + """ + 舊協定:先在第一段做驗證,再靠括號平衡判斷何時收完。 + Legacy path: authorise on the first segment, then use bracket balance + to decide when the body has fully arrived. + + Authorisation cannot wait for a complete body here — a client sending + a bad or absent token must be rejected immediately rather than kept + waiting for bytes it is never going to send. + """ + payload = self._authorize(bytes(buffer).strip()) if payload is None: + return None + scanner = _RequestFrameScanner() + scanner.feed(payload) + body = bytearray(payload) + # Not ``_fill_until``: the scanner has to be fed every chunk, so the + # loop advances both the buffer and the stop condition together. + while not scanner.is_complete: + if len(body) > _MAX_REQUEST_BYTES: + print(f"request exceeded {_MAX_REQUEST_BYTES} bytes; dropping", flush=True) + return None + chunk = self._recv_chunk() + if not chunk: + # A peer that closes early still gets its reply: whatever + # arrived is handed on so a parse error comes back rather + # than silence. + break + body.extend(chunk) + scanner.feed(chunk) + return bytes(body) + + def _receive_request(self) -> bytes | None: + """ + 讀滿一個完整請求,自動辨識長度前綴或舊格式。 + Read one whole request, auto-detecting framed vs legacy format. + """ + self.request.settimeout(_RECV_TIMEOUT_SECONDS) + buffer = bytearray() + mode = _MODE_UNDECIDED + while mode == _MODE_UNDECIDED: + chunk = self._recv_chunk() + if not chunk: + return None + buffer.extend(chunk) + mode = _detect_mode(bytes(buffer)) + if mode == _MODE_FRAMED: + self._framed = True + body = self._read_framed_body(buffer) + # The frame is transport only; auth still applies to its contents, + # and now sees the whole body at once. + return None if body is None else self._authorize(body.strip()) + return self._read_legacy_body(buffer) + + def handle(self): + self._framed = False + raw = self._receive_request() + if raw is None: return - command_string = payload.decode("utf-8", errors="replace").strip() + command_string = raw.decode("utf-8", errors="replace").strip() print("command is: " + command_string, flush=True) if command_string == _QUIT_COMMAND: self._handle_quit() @@ -97,6 +355,69 @@ def __init__(self, server_address, request_handler_class, auth_token: str | None self.auth_token: str | None = auth_token +def _read_exactly(sock, count: int) -> bytes: + """Read exactly ``count`` bytes or raise on a short stream.""" + buffer = bytearray() + while len(buffer) < count: + chunk = sock.recv(min(_RECV_BYTES, count - len(buffer))) + if not chunk: + raise ConnectionError( + f"connection closed after {len(buffer)} of {count} bytes" + ) + buffer.extend(chunk) + return bytes(buffer) + + +def read_frame(sock) -> bytes: + """ + 從 ``sock`` 讀取一個長度前綴訊框並回傳其內容。 + Read one length-prefixed frame from ``sock`` and return its body. + + :param sock: 已連線的 socket / a connected socket + :return: 訊框內容 / the frame body + """ + header = bytearray() + while not header.endswith(_NEWLINE): + chunk = sock.recv(1) + if not chunk: + raise ConnectionError("connection closed while reading frame header") + header.extend(chunk) + if len(header) > _MAX_HEADER_BYTES: + raise ValueError(f"frame header too long: {bytes(header)!r}") + if not header.startswith(_LENGTH_PREFIX): + raise ValueError(f"not a length-prefixed frame: {bytes(header)!r}") + declared = int(bytes(header)[len(_LENGTH_PREFIX):].strip()) + return _read_exactly(sock, declared) + + +def send_command( + sock, + command: str | bytes, + auth_token: str | None = None, +) -> bytes: + """ + 以長度前綴格式送出一筆指令並讀回回覆。 + Send one command using length-prefixed framing and return the reply body. + + Framing removes the guesswork of the legacy protocol: neither side has to + infer where a message ends, so oversized payloads and a token line landing + on a packet boundary are both handled correctly. + + :param sock: 已連線到 WebRunner socket server 的 socket + A socket already connected to the WebRunner socket server + :param command: 動作 JSON 字串或 ``quit_server`` + Action JSON string, or ``quit_server`` + :param auth_token: 伺服器啟用 token 驗證時必填 / required when the server + was started with ``auth_token`` + :return: 伺服器回覆內容 / the server's reply body + """ + body = command.encode("utf-8") if isinstance(command, str) else bytes(command) + if auth_token is not None: + body = auth_token.encode("utf-8") + _NEWLINE + body + sock.sendall(encode_frame(body)) + return read_frame(sock) + + def _build_tls_context(certfile: str, keyfile: str) -> ssl.SSLContext: # ``Purpose.CLIENT_AUTH`` only sets the side; explicitly pin TLSv1.2+ # so SonarCloud S4423 is satisfied even on older Pythons that may diff --git a/test/unit_test/test_socket_server.py b/test/unit_test/test_socket_server.py index e16c9916..6acc2ef9 100644 --- a/test/unit_test/test_socket_server.py +++ b/test/unit_test/test_socket_server.py @@ -1,9 +1,18 @@ +import json import socket import threading -import time import unittest from je_web_runner.utils.socket_server.web_runner_socket_server import ( + _detect_mode, + _MODE_FRAMED, + _MODE_LEGACY, + _MODE_UNDECIDED, + _RECV_BYTES, + _RequestFrameScanner, + encode_frame, + read_frame, + send_command, TCPServer, TCPServerHandler, ) @@ -86,5 +95,249 @@ def test_quit_with_correct_token_shuts_down(self): pass +class TestRequestFrameScanner(unittest.TestCase): + """The stop condition that decides whether more bytes are still coming.""" + + @staticmethod + def _scan(*chunks): + scanner = _RequestFrameScanner() + for chunk in chunks: + scanner.feed(chunk) + return scanner.is_complete + + def test_balanced_document_is_complete(self): + self.assertTrue(self._scan(b'[["WR_quit"]]')) + + def test_bare_command_is_complete(self): + self.assertTrue(self._scan(b"quit_server")) + + def test_truncated_document_is_incomplete(self): + self.assertFalse(self._scan(b'[["WR_quit"')) + + def test_unterminated_string_is_incomplete(self): + self.assertFalse(self._scan(b'[["WR_qu')) + + def test_brackets_inside_string_do_not_count(self): + # The ']' characters are string content, not structure. + self.assertFalse(self._scan(b'[["]]]]"')) + self.assertTrue(self._scan(b'[["]]]]"]]')) + + def test_escaped_quote_does_not_close_string(self): + self.assertFalse(self._scan(b'[["a\\""')) + self.assertTrue(self._scan(b'[["a\\""]]')) + + def test_completeness_is_reached_across_chunks(self): + self.assertTrue(self._scan(b'[["WR_', b'quit"]', b"]")) + + def test_empty_input_is_incomplete(self): + self.assertFalse(self._scan(b"", b" \n ")) + + def test_excess_closer_stops_reading(self): + # Malformed rather than truncated — waiting for more would just + # stall until the timeout, so treat it as finished and let the + # JSON parse fail into an error reply. + self.assertTrue(self._scan(b'[["a"]]]]')) + + def test_multibyte_char_split_across_chunks(self): + payload = '[["測試"]]'.encode("utf-8") + # Split mid-character; the incremental decoder must not mangle it + # into something that looks like a delimiter. + self.assertTrue(self._scan(payload[:6], payload[6:])) + + +class TestLargeAndFragmentedRequests(unittest.TestCase): + """TCP is a stream: one recv() is not one message.""" + + def setUp(self): + self.server, _ = _start_server() + self.addCleanup(self._stop) + + def _stop(self): + try: + self.server.shutdown() + self.server.server_close() + except OSError: + pass + + def _round_trip(self, payload: bytes, chunk_size: int | None = None) -> bytes: + host, port = self.server.server_address + with socket.create_connection((host, port)) as client: + if chunk_size is None: + client.sendall(payload) + else: + for start in range(0, len(payload), chunk_size): + client.sendall(payload[start:start + chunk_size]) + return _read_until_marker(client) + + def test_payload_larger_than_one_recv_is_not_truncated(self): + # Comfortably more than a single _RECV_BYTES read. + padding = "A" * (_RECV_BYTES * 6) + payload = json.dumps([["WR_unknown_command", {"padding": padding}]]).encode() + self.assertGreater(len(payload), _RECV_BYTES) + + reply = self._round_trip(payload) + + # Reaching the executor at all proves the whole document was parsed: + # a truncated body would have failed in json.loads instead. + self.assertIn(b"unknown command: WR_unknown_command", reply) + + def test_fragmented_send_is_reassembled(self): + payload = json.dumps([["WR_unknown_command", {"padding": "B" * 20000}]]).encode() + + reply = self._round_trip(payload, chunk_size=1024) + + self.assertIn(b"unknown command: WR_unknown_command", reply) + + def test_malformed_but_balanced_json_replies_promptly(self): + # Structure is closed, so the server must stop reading and report the + # parse error rather than blocking until the receive timeout. + reply = self._round_trip(b"[[[not valid json]]]") + self.assertIn(b"Return_Data_Over_JE", reply) + self.assertNotIn(b"unknown command", reply) + + +class TestFrameEncoding(unittest.TestCase): + + def test_encode_frame_layout(self): + self.assertEqual(encode_frame(b"hello"), b"WRLEN 5\nhello") + + def test_encode_empty_body(self): + self.assertEqual(encode_frame(b""), b"WRLEN 0\n") + + def test_length_counts_bytes_not_characters(self): + body = "測試".encode("utf-8") # 2 chars, 6 bytes + self.assertEqual(encode_frame(body), b"WRLEN 6\n" + body) + + def test_non_bytes_rejected(self): + with self.assertRaises(TypeError): + encode_frame("not bytes") # type: ignore[arg-type] + + +class TestDetectMode(unittest.TestCase): + """Auto-detection is what keeps legacy clients working untouched.""" + + def test_marker_selects_framed(self): + self.assertEqual(_detect_mode(b"WRLEN 12\n"), _MODE_FRAMED) + + def test_json_selects_legacy(self): + self.assertEqual(_detect_mode(b'[["WR_quit"]]'), _MODE_LEGACY) + + def test_partial_marker_is_undecided(self): + # A tiny first recv must not be mistaken for a legacy request. + for size in range(1, len(b"WRLEN ")): + self.assertEqual(_detect_mode(b"WRLEN "[:size]), _MODE_UNDECIDED) + + def test_empty_is_undecided(self): + self.assertEqual(_detect_mode(b""), _MODE_UNDECIDED) + + def test_near_miss_is_legacy(self): + self.assertEqual(_detect_mode(b"WRLENX"), _MODE_LEGACY) + self.assertEqual(_detect_mode(b"wrlen "), _MODE_LEGACY) + + +class TestFramedRequests(unittest.TestCase): + + def _serve(self, auth_token=None): + server, _ = _start_server(auth_token=auth_token) + self.addCleanup(self._stop, server) + return server + + @staticmethod + def _stop(server): + try: + server.shutdown() + server.server_close() + except OSError: + pass + + def test_framed_round_trip(self): + server = self._serve() + host, port = server.server_address + with socket.create_connection((host, port)) as client: + reply = send_command(client, json.dumps([["WR_unknown_command"]])) + self.assertIn(b"unknown command: WR_unknown_command", reply) + + def test_framed_payload_larger_than_one_recv(self): + server = self._serve() + host, port = server.server_address + padding = "A" * (_RECV_BYTES * 6) + command = json.dumps([["WR_unknown_command", {"padding": padding}]]) + with socket.create_connection((host, port)) as client: + reply = send_command(client, command) + self.assertIn(b"unknown command: WR_unknown_command", reply) + + def test_framed_token_split_across_packets_is_authorised(self): + """The reason for the length prefix: with an explicit byte count the + token line no longer has to arrive inside the first recv.""" + server = self._serve(auth_token="secret") # nosec B106 — fake fixture + host, port = server.server_address + body = b"secret\n" + json.dumps([["WR_unknown_command"]]).encode() + frame = encode_frame(body) + with socket.create_connection((host, port)) as client: + # Dribble the header and token out one byte at a time so the + # token straddles several recv() calls. + for index in range(14): + client.sendall(frame[index:index + 1]) + client.sendall(frame[14:]) + reply = read_frame(client) + self.assertIn(b"unknown command: WR_unknown_command", reply) + self.assertNotIn(b"unauthorized", reply) + + def test_framed_wrong_token_is_rejected(self): + server = self._serve(auth_token="secret") # nosec B106 — fake fixture + host, port = server.server_address + with socket.create_connection((host, port)) as client: + reply = send_command( + client, json.dumps([["WR_unknown_command"]]), auth_token="wrong", + ) + self.assertIn(b"unauthorized", reply) + self.assertFalse(server.close_flag) + + def test_framed_quit_acknowledges_then_shuts_down(self): + server = self._serve() + host, port = server.server_address + with socket.create_connection((host, port)) as client: + reply = send_command(client, "quit_server") + self.assertIn(b"quit_server", reply) + self.assertTrue(server.close_event.wait(timeout=2.0)) + + def test_under_length_body_is_not_executed(self): + """A declared length the peer never delivers is a truncated request, + so it must not be handed to the executor.""" + server = self._serve() + host, port = server.server_address + command = json.dumps([["WR_unknown_command"]]).encode() + with socket.create_connection((host, port)) as client: + # Claim more bytes than are actually sent, then close. + client.sendall(b"WRLEN " + str(len(command) + 500).encode() + b"\n") + client.sendall(command) + client.shutdown(socket.SHUT_WR) + leftover = client.recv(4096) + self.assertEqual(leftover, b"") + + def test_malformed_length_header_is_dropped(self): + server = self._serve() + host, port = server.server_address + with socket.create_connection((host, port)) as client: + client.sendall(b"WRLEN not-a-number\n[]") + client.shutdown(socket.SHUT_WR) + leftover = client.recv(4096) + self.assertEqual(leftover, b"") + + +class TestReadFrame(unittest.TestCase): + + def test_rejects_non_frame(self): + server, _ = _start_server() + self.addCleanup(TestFramedRequests._stop, server) + host, port = server.server_address + # The server replies to a legacy request with the sentinel format, + # which read_frame must refuse rather than misparse. + with socket.create_connection((host, port)) as client: + client.sendall(json.dumps([["WR_unknown_command"]]).encode()) + with self.assertRaises(ValueError): + read_frame(client) + + if __name__ == "__main__": unittest.main() From 6b9d5ccf3e111dca301ab4812df4ae3af209f9a9 Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Thu, 23 Jul 2026 11:20:47 +0800 Subject: [PATCH 5/8] Surface non-comparable pagination keys as audit errors items_by_page_key only guarantees Hashable, which does not imply orderable. Mixed key types raised a bare TypeError out of assert_sorted_by instead of the module exception callers handle. --- je_web_runner/utils/pagination_audit/audit.py | 27 ++++++++++++------- test/unit_test/test_pagination_audit.py | 13 +++++++++ 2 files changed, 30 insertions(+), 10 deletions(-) diff --git a/je_web_runner/utils/pagination_audit/audit.py b/je_web_runner/utils/pagination_audit/audit.py index 5d15c9c4..869fa197 100644 --- a/je_web_runner/utils/pagination_audit/audit.py +++ b/je_web_runner/utils/pagination_audit/audit.py @@ -223,14 +223,21 @@ def assert_sorted_by( return last = flattened[0] for current in flattened[1:]: - if reverse: - if current > last: - raise PaginationAuditError( - f"order violation: {current!r} > {last!r} but reverse=True" - ) - else: - if current < last: - raise PaginationAuditError( - f"order violation: {current!r} < {last!r}" - ) + # ``items_by_page_key`` is caller-supplied and only guaranteed to + # return Hashable, which is not necessarily orderable. Mixed types + # (str vs int, None vs anything) raise TypeError here — surface that + # as a PaginationAuditError instead of leaking a bare TypeError. + try: + out_of_order = current > last if reverse else current < last + except TypeError as error: + raise PaginationAuditError( + f"items_by_page_key returned non-comparable keys " + f"({current!r} vs {last!r}): {error}" + ) from error + if out_of_order: + symbol = ">" if reverse else "<" + suffix = " but reverse=True" if reverse else "" + raise PaginationAuditError( + f"order violation: {current!r} {symbol} {last!r}{suffix}" + ) last = current diff --git a/test/unit_test/test_pagination_audit.py b/test/unit_test/test_pagination_audit.py index 41b3cda4..0bc3a192 100644 --- a/test/unit_test/test_pagination_audit.py +++ b/test/unit_test/test_pagination_audit.py @@ -207,6 +207,19 @@ def _boom(_key): with self.assertRaises(PaginationAuditError): assert_sorted_by(findings, _boom) + def test_non_comparable_keys_raise_audit_error(self): + # Hashable does not imply orderable: mixing str and int makes the + # ``<`` comparison raise TypeError. That must surface as a + # PaginationAuditError, not leak a bare TypeError to the caller. + findings = PaginationFindings(item_keys_by_page=[[1, "two"]]) + with self.assertRaises(PaginationAuditError): + assert_sorted_by(findings, lambda x: x) + + def test_none_key_raises_audit_error(self): + findings = PaginationFindings(item_keys_by_page=[[1], [None]]) + with self.assertRaises(PaginationAuditError): + assert_sorted_by(findings, lambda x: x) + if __name__ == "__main__": unittest.main() From 5ef4b65e494b4456a163f804163b5b4aaebfe06e Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Thu, 23 Jul 2026 11:20:47 +0800 Subject: [PATCH 6/8] Make the fan-out parallelism test deterministic The 0.12s wall-clock budget was flaky on loaded machines, where thread start-up alone can exceed it. Use a barrier instead: three tasks can only clear it when they are genuinely in flight at the same time. --- test/unit_test/test_fanout.py | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/test/unit_test/test_fanout.py b/test/unit_test/test_fanout.py index 2e3f72c0..46df9e55 100644 --- a/test/unit_test/test_fanout.py +++ b/test/unit_test/test_fanout.py @@ -1,4 +1,4 @@ -import time +import threading import unittest from je_web_runner.utils.fanout import ( @@ -54,15 +54,19 @@ def test_invalid_task_raises(self): run_fan_out([42]) # type: ignore[list-item] def test_actually_runs_in_parallel(self): - def slow(): - time.sleep(0.05) + # Proving concurrency with a wall-clock budget is flaky on loaded CI + # boxes (thread start-up alone can exceed the margin). A barrier is + # deterministic: all three tasks can only clear it if they are in + # flight at the same time, otherwise wait() raises BrokenBarrierError. + barrier = threading.Barrier(3, timeout=10) + + def rendezvous(): + barrier.wait() return "ok" - start = time.monotonic() - result = run_fan_out([slow, slow, slow], max_workers=3) - elapsed = time.monotonic() - start - # Sequential would be ~0.15s; parallel should land well under 0.12s. - self.assertTrue(result.succeeded) - self.assertLess(elapsed, 0.12) + + result = run_fan_out([rendezvous, rendezvous, rendezvous], max_workers=3) + self.assertTrue(result.succeeded, [o.to_dict() for o in result.failures]) + self.assertEqual(len(result.outcomes), 3) if __name__ == "__main__": From cf8d891402d5d09d66c80714bbf6e80dd38e6066 Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Thu, 23 Jul 2026 11:27:43 +0800 Subject: [PATCH 7/8] Mark the invalid-token test fixture as a deliberate non-secret Bandit B106 flagged the intentionally wrong token used to assert that authentication rejects it, matching the suppression the surrounding fixtures already carry. --- test/unit_test/test_socket_server.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/unit_test/test_socket_server.py b/test/unit_test/test_socket_server.py index 6acc2ef9..f877e066 100644 --- a/test/unit_test/test_socket_server.py +++ b/test/unit_test/test_socket_server.py @@ -287,7 +287,7 @@ def test_framed_wrong_token_is_rejected(self): server = self._serve(auth_token="secret") # nosec B106 — fake fixture host, port = server.server_address with socket.create_connection((host, port)) as client: - reply = send_command( + reply = send_command( # nosec B106 — deliberately invalid fixture token client, json.dumps([["WR_unknown_command"]]), auth_token="wrong", ) self.assertIn(b"unauthorized", reply) From 5ea57fb188bd294911a25235089c4e30774c5838 Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Thu, 23 Jul 2026 12:56:06 +0800 Subject: [PATCH 8/8] Annotate remaining Bandit hardcoded-secret false positives All six were fixture data or protocol constants rather than credentials: stubbed OAuth/FCM tokens, the "Bearer" token-type constant (the mock server issues its real token from secrets.token_hex), and "pass_rate", which matched only on the "pass" substring and holds a float ratio. Bandit parses space-separated test IDs after nosec, not comma-separated, and the marker has to sit on the flagged line itself. Both tripped up the existing suppression on the autofill fixture. Bandit now reports clean across the package and the test suite at every severity level, not just the -ll threshold CI had been using. --- je_web_runner/utils/mock_services/servers.py | 2 ++ test/unit_test/test_form_autofill.py | 2 +- test/unit_test/test_oauth.py | 6 +++--- test/unit_test/test_oauth_pkce_replay.py | 3 ++- test/unit_test/test_push_delivery.py | 2 +- test/unit_test/test_trend_dashboard.py | 2 +- 6 files changed, 10 insertions(+), 7 deletions(-) diff --git a/je_web_runner/utils/mock_services/servers.py b/je_web_runner/utils/mock_services/servers.py index 2e58f816..887e9388 100644 --- a/je_web_runner/utils/mock_services/servers.py +++ b/je_web_runner/utils/mock_services/servers.py @@ -153,6 +153,8 @@ def do_POST(self): server_state["issued"].append(token) self._send(200, { "access_token": token, + # nosec B105 — OAuth token *type* constant; the token itself + # above comes from secrets.token_hex. "token_type": "Bearer", "expires_in": 3600, }) diff --git a/test/unit_test/test_form_autofill.py b/test/unit_test/test_form_autofill.py index bc60e387..7585d8ed 100644 --- a/test/unit_test/test_form_autofill.py +++ b/test/unit_test/test_form_autofill.py @@ -80,7 +80,7 @@ def test_generates_action_triplet(self): {"type": "email", "id": "email", "label": "Email"}, {"type": "password", "id": "pwd", "label": "Password"}, ] - fixture = {"email": "a@b.com", "password": "wonder"} # NOSONAR # nosec B106 — fake fixture + fixture = {"email": "a@b.com", "password": "wonder"} # NOSONAR # nosec B105 B106 — fake fixture actions = plan_fill_actions(fields, fixture) commands = [a[0] for a in actions] # Three-step block per field: save_test_object, find, input diff --git a/test/unit_test/test_oauth.py b/test/unit_test/test_oauth.py index 6279a0c0..f4f8fef2 100644 --- a/test/unit_test/test_oauth.py +++ b/test/unit_test/test_oauth.py @@ -14,8 +14,8 @@ def _success_response(**overrides): response = MagicMock(status_code=200, text="ok") payload = { - "access_token": "abc", - "token_type": "Bearer", + "access_token": "abc", # nosec B105 — fake token in a stubbed OAuth response + "token_type": "Bearer", # nosec B105 — OAuth token *type*, not a credential "expires_in": 3600, } payload.update(overrides) @@ -30,7 +30,7 @@ def setUp(self): def test_invalid_url_raises(self): with self.assertRaises(OAuthError): - client_credentials_token("ftp://example.com", "id", "secret") # NOSONAR — fixture, asserts the validator rejects it + client_credentials_token("ftp://example.com", "id", "secret") # NOSONAR — fixture def test_returns_token_response(self): with patch("je_web_runner.utils.auth.oauth.requests.post", diff --git a/test/unit_test/test_oauth_pkce_replay.py b/test/unit_test/test_oauth_pkce_replay.py index 0403d6d8..3a47f7da 100644 --- a/test/unit_test/test_oauth_pkce_replay.py +++ b/test/unit_test/test_oauth_pkce_replay.py @@ -52,7 +52,8 @@ def probe(payload): def test_accepted_outcome_is_bug(self): def probe(payload): return TokenExchangeResponse( - status_code=200, body={"access_token": "abc"}, + status_code=200, + body={"access_token": "abc"}, # nosec B105 — fake stubbed token ) result = replay(ReplayCase(name="x", payload={}), probe) self.assertEqual(result.outcome, ReplayOutcome.ACCEPTED) diff --git a/test/unit_test/test_push_delivery.py b/test/unit_test/test_push_delivery.py index dbcd97cc..db53cac3 100644 --- a/test/unit_test/test_push_delivery.py +++ b/test/unit_test/test_push_delivery.py @@ -12,7 +12,7 @@ def _good_fcm(): return { "message": { - "token": "device-token", + "token": "device-token", # nosec B105 — fake FCM device token fixture "notification": {"title": "T", "body": "B"}, "android": {"ttl": "3600s"}, }, diff --git a/test/unit_test/test_trend_dashboard.py b/test/unit_test/test_trend_dashboard.py index b277cfce..bab0ee70 100644 --- a/test/unit_test/test_trend_dashboard.py +++ b/test/unit_test/test_trend_dashboard.py @@ -60,7 +60,7 @@ class TestRender(unittest.TestCase): def test_render_includes_title_and_table(self): trend = {"daily": [ {"label": "2026-04-25", "passed": 1, "failed": 0, "total": 1, - "pass_rate": 1.0, "avg_duration_seconds": 1.5}, + "pass_rate": 1.0, "avg_duration_seconds": 1.5}, # nosec B105 — a ratio, not a secret ], "totals": {}} text = render_html(trend, title="Demo") self.assertIn("Demo", text)